#coding: utf-8

'''

The idea is that we record the commands sent to the debugger and reproduce them from this script

(so, this works as the client, which spawns the debugger as a separate process and communicates

to it as if it was run from the outside)

Note that it's a python script but it'll spawn a process to run as jython, ironpython and as python.

'''

import os

import platform

import sys

import threading

import time

import unittest

import pytest

from tests_python import debugger_unittest

from tests_python.debugger_unittest import (get_free_port, CMD_SET_PROPERTY_TRACE, REASON_CAUGHT_EXCEPTION,

REASON_UNCAUGHT_EXCEPTION, REASON_STOP_ON_BREAKPOINT, REASON_THREAD_SUSPEND, overrides, CMD_THREAD_CREATE,

CMD_GET_THREAD_STACK, REASON_STEP_INTO_MY_CODE, CMD_GET_EXCEPTION_DETAILS)

from _pydevd_bundle.pydevd_constants import IS_WINDOWS

try:

from urllib import unquote

except ImportError:

from urllib.parse import unquote

IS_CPYTHON = platform.python_implementation() == 'CPython'

IS_IRONPYTHON = platform.python_implementation() == 'IronPython'

IS_JYTHON = platform.python_implementation() == 'Jython'

IS_APPVEYOR = os.environ.get('APPVEYOR', '') in ('True', 'true', '1')

try:

xrange

except:

xrange = range

TEST_DJANGO = False

if sys.version_info[:2] == (2, 7):

# Only test on python 2.7 for now

try:

import django

TEST_DJANGO = True

except:

pass

IS_PY2 = False

if sys.version_info[0] == 2:

IS_PY2 = True

IS_PY26 = sys.version_info[:2] == (2, 6)

if IS_PY2:

builtin_qualifier = "__builtin__"

else:

builtin_qualifier = "builtins"

IS_PY36 = False

if sys.version_info[0] == 3 and sys.version_info[1] == 6:

IS_PY36 = True

from tests_python.debug_constants import TEST_CYTHON

from tests_python.debug_constants import TEST_JYTHON

#=======================================================================================================================

# WriterThreadCaseSetNextStatement

#======================================================================================================================

class WriterThreadCaseSetNextStatement(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case_set_next_statement.py')

def run(self):

self.start_socket()

breakpoint_id = self.write_add_breakpoint(6, None)

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=6)

self.write_evaluate_expression('%s\t%s\t%s' % (hit.thread_id, hit.frame_id, 'LOCAL'), 'a')

self.wait_for_evaluation('<var name="a" type="int" qualifier="{0}" value="int: 2"'.format(builtin_qualifier))

self.write_set_next_statement(hit.thread_id, 2, 'method')

hit = self.wait_for_breakpoint_hit('127', line=2)

self.write_step_over(hit.thread_id)

hit = self.wait_for_breakpoint_hit('108')

self.write_evaluate_expression('%s\t%s\t%s' % (hit.thread_id, hit.frame_id, 'LOCAL'), 'a')

self.wait_for_evaluation('<var name="a" type="int" qualifier="{0}" value="int: 1"'.format(builtin_qualifier))

self.write_remove_breakpoint(breakpoint_id)

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCaseGetNextStatementTargets

#======================================================================================================================

class WriterThreadCaseGetNextStatementTargets(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case_get_next_statement_targets.py')

def run(self):

self.start_socket()

breakpoint_id = self.write_add_breakpoint(21, None)

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=21)

self.write_get_next_statement_targets(hit.thread_id, hit.frame_id)

targets = self.wait_for_get_next_statement_targets()

expected = set((2, 3, 5, 8, 9, 10, 12, 13, 14, 15, 17, 18, 19, 21))

assert targets == expected, 'Expected targets to be %s, was: %s' % (expected, targets)

self.write_remove_breakpoint(breakpoint_id)

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# AbstractWriterThreadCaseDjango

#======================================================================================================================

class AbstractWriterThreadCaseDjango(debugger_unittest.AbstractWriterThread):

FORCE_KILL_PROCESS_WHEN_FINISHED_OK = True

def _ignore_stderr_line(self, line):

if debugger_unittest.AbstractWriterThread._ignore_stderr_line(self, line):

return True

if 'GET /my_app' in line:

return True

return False

def get_command_line_args(self):

free_port = get_free_port()

self.django_port = free_port

return [

debugger_unittest._get_debugger_test_file(os.path.join('my_django_proj_17', 'manage.py')),

'runserver',

'--noreload',

str(free_port),

]

def write_add_breakpoint_django(self, line, func, template):

'''

@param line: starts at 1

'''

breakpoint_id = self.next_breakpoint_id()

template_file = debugger_unittest._get_debugger_test_file(os.path.join('my_django_proj_17', 'my_app', 'templates', 'my_app', template))

self.write("111\t%s\t%s\t%s\t%s\t%s\t%s\tNone\tNone" % (self.next_seq(), breakpoint_id, 'django-line', template_file, line, func))

self.log.append('write_add_django_breakpoint: %s line: %s func: %s' % (breakpoint_id, line, func))

return breakpoint_id

def create_request_thread(self, uri):

outer= self

class T(threading.Thread):

def run(self):

try:

from urllib.request import urlopen

except ImportError:

from urllib import urlopen

for _ in xrange(10):

try:

stream = urlopen('http://127.0.0.1:%s/%s' % (outer.django_port,uri))

self.contents = stream.read()

break

except IOError:

continue

return T()

#=======================================================================================================================

# WriterThreadCaseDjango

#======================================================================================================================

class WriterThreadCaseDjango(AbstractWriterThreadCaseDjango):

def run(self):

self.start_socket()

self.write_add_breakpoint_django(5, None, 'index.html')

self.write_make_initial_run()

t = self.create_request_thread('my_app')

time.sleep(5) # Give django some time to get to startup before requesting the page

t.start()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=5)

self.write_get_variable(hit.thread_id, hit.frame_id, 'entry')

self.wait_for_vars([

'<var name="key" type="str"',

'v1'

])

self.write_run_thread(hit.thread_id)

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=5)

self.write_get_variable(hit.thread_id, hit.frame_id, 'entry')

self.wait_for_vars([

'<var name="key" type="str"',

'v2'

])

self.write_run_thread(hit.thread_id)

for _ in xrange(10):

if hasattr(t, 'contents'):

break

time.sleep(.3)

else:

raise AssertionError('Django did not return contents properly!')

contents = t.contents.replace(' ', '').replace('\r', '').replace('\n', '')

if contents != '<ul><li>v1:v1</li><li>v2:v2</li></ul>':

raise AssertionError('%s != <ul><li>v1:v1</li><li>v2:v2</li></ul>' % (contents,))

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCaseDjango2

#======================================================================================================================

class WriterThreadCaseDjango2(AbstractWriterThreadCaseDjango):

def run(self):

self.start_socket()

self.write_add_breakpoint_django(4, None, 'name.html')

self.write_make_initial_run()

t = self.create_request_thread('my_app/name')

time.sleep(5) # Give django some time to get to startup before requesting the page

t.start()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=4)

self.write_get_frame(hit.thread_id, hit.frame_id)

self.wait_for_var('<var name="form" type="NameForm" qualifier="my_app.forms" value="NameForm%253A')

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase19 - [Test Case]: Evaluate '__' attributes

#======================================================================================================================

class WriterThreadCase19(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case19.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(8, None)

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=8)

self.write_evaluate_expression('%s\t%s\t%s' % (hit.thread_id, hit.frame_id, 'LOCAL'), 'a.__var')

self.wait_for_evaluation([

[

'<var name="a.__var" type="int" qualifier="{0}" value="int'.format(builtin_qualifier),

'<var name="a.__var" type="int" value="int', # jython

]

])

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase20 - Check that we were notified of threads creation before they started to run

#======================================================================================================================

class WriterThreadCase20(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case20.py')

def run(self):

self.start_socket()

self.write_make_initial_run()

# We already check if it prints 'TEST SUCEEDED' by default, so, nothing

# else should be needed in this test as it tests what's needed just by

# running the module.

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase18 - [Test Case]: change local variable

#======================================================================================================================

class WriterThreadCase18(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case18.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(5, 'm2')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=5)

self.write_change_variable(hit.thread_id, hit.frame_id, 'a', '40')

self.wait_for_var('<xml><var name="" type="int" qualifier="{0}" value="int%253A 40" />%0A</xml>'.format(builtin_qualifier,))

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase17 - [Test Case]: dont trace

#======================================================================================================================

class WriterThreadCase17(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case17.py')

def run(self):

self.start_socket()

self.write_enable_dont_trace(True)

self.write_add_breakpoint(27, 'main')

self.write_add_breakpoint(29, 'main')

self.write_add_breakpoint(31, 'main')

self.write_add_breakpoint(33, 'main')

self.write_make_initial_run()

for _i in range(4):

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT)

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107', line=2)

# Should Skip step into properties setter

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase17a - [Test Case]: dont trace return

#======================================================================================================================

class WriterThreadCase17a(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case17a.py')

def run(self):

self.start_socket()

self.write_enable_dont_trace(True)

self.write_add_breakpoint(2, 'm1')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT, line=2)

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107', line=10)

# Should Skip step into properties setter

assert hit.name == 'm3'

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase16 - [Test Case]: numpy.ndarray resolver

#======================================================================================================================

class WriterThreadCase16(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case16.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(9, 'main')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT)

# In this test we check that the three arrays of different shapes, sizes and types

# are all resolved properly as ndarrays.

# First pass check is that we have all three expected variables defined

self.write_get_frame(hit.thread_id, hit.frame_id)

self.wait_for_multiple_vars((

(

'<var name="smallarray" type="ndarray" qualifier="numpy" value="ndarray%253A %255B 0.%252B1.j 1.%252B1.j 2.%252B1.j 3.%252B1.j 4.%252B1.j 5.%252B1.j 6.%252B1.j 7.%252B1.j 8.%252B1.j%250A 9.%252B1.j 10.%252B1.j 11.%252B1.j 12.%252B1.j 13.%252B1.j 14.%252B1.j 15.%252B1.j 16.%252B1.j 17.%252B1.j%250A 18.%252B1.j 19.%252B1.j 20.%252B1.j 21.%252B1.j 22.%252B1.j 23.%252B1.j 24.%252B1.j 25.%252B1.j 26.%252B1.j%250A 27.%252B1.j 28.%252B1.j 29.%252B1.j 30.%252B1.j 31.%252B1.j 32.%252B1.j 33.%252B1.j 34.%252B1.j 35.%252B1.j%250A 36.%252B1.j 37.%252B1.j 38.%252B1.j 39.%252B1.j 40.%252B1.j 41.%252B1.j 42.%252B1.j 43.%252B1.j 44.%252B1.j%250A 45.%252B1.j 46.%252B1.j 47.%252B1.j 48.%252B1.j 49.%252B1.j 50.%252B1.j 51.%252B1.j 52.%252B1.j 53.%252B1.j%250A 54.%252B1.j 55.%252B1.j 56.%252B1.j 57.%252B1.j 58.%252B1.j 59.%252B1.j 60.%252B1.j 61.%252B1.j 62.%252B1.j%250A 63.%252B1.j 64.%252B1.j 65.%252B1.j 66.%252B1.j 67.%252B1.j 68.%252B1.j 69.%252B1.j 70.%252B1.j 71.%252B1.j%250A 72.%252B1.j 73.%252B1.j 74.%252B1.j 75.%252B1.j 76.%252B1.j 77.%252B1.j 78.%252B1.j 79.%252B1.j 80.%252B1.j%250A 81.%252B1.j 82.%252B1.j 83.%252B1.j 84.%252B1.j 85.%252B1.j 86.%252B1.j 87.%252B1.j 88.%252B1.j 89.%252B1.j%250A 90.%252B1.j 91.%252B1.j 92.%252B1.j 93.%252B1.j 94.%252B1.j 95.%252B1.j 96.%252B1.j 97.%252B1.j 98.%252B1.j%250A 99.%252B1.j%255D" isContainer="True" />',

'<var name="smallarray" type="ndarray" qualifier="numpy" value="ndarray%253A %255B 0.%252B1.j 1.%252B1.j 2.%252B1.j 3.%252B1.j 4.%252B1.j 5.%252B1.j 6.%252B1.j 7.%252B1.j%250A 8.%252B1.j 9.%252B1.j 10.%252B1.j 11.%252B1.j 12.%252B1.j 13.%252B1.j 14.%252B1.j 15.%252B1.j%250A 16.%252B1.j 17.%252B1.j 18.%252B1.j 19.%252B1.j 20.%252B1.j 21.%252B1.j 22.%252B1.j 23.%252B1.j%250A 24.%252B1.j 25.%252B1.j 26.%252B1.j 27.%252B1.j 28.%252B1.j 29.%252B1.j 30.%252B1.j 31.%252B1.j%250A 32.%252B1.j 33.%252B1.j 34.%252B1.j 35.%252B1.j 36.%252B1.j 37.%252B1.j 38.%252B1.j 39.%252B1.j%250A 40.%252B1.j 41.%252B1.j 42.%252B1.j 43.%252B1.j 44.%252B1.j 45.%252B1.j 46.%252B1.j 47.%252B1.j%250A 48.%252B1.j 49.%252B1.j 50.%252B1.j 51.%252B1.j 52.%252B1.j 53.%252B1.j 54.%252B1.j 55.%252B1.j%250A 56.%252B1.j 57.%252B1.j 58.%252B1.j 59.%252B1.j 60.%252B1.j 61.%252B1.j 62.%252B1.j 63.%252B1.j%250A 64.%252B1.j 65.%252B1.j 66.%252B1.j 67.%252B1.j 68.%252B1.j 69.%252B1.j 70.%252B1.j 71.%252B1.j%250A 72.%252B1.j 73.%252B1.j 74.%252B1.j 75.%252B1.j 76.%252B1.j 77.%252B1.j 78.%252B1.j 79.%252B1.j%250A 80.%252B1.j 81.%252B1.j 82.%252B1.j 83.%252B1.j 84.%252B1.j 85.%252B1.j 86.%252B1.j 87.%252B1.j%250A 88.%252B1.j 89.%252B1.j 90.%252B1.j 91.%252B1.j 92.%252B1.j 93.%252B1.j 94.%252B1.j 95.%252B1.j%250A 96.%252B1.j 97.%252B1.j 98.%252B1.j 99.%252B1.j%255D" isContainer="True" />'

),

(

'<var name="bigarray" type="ndarray" qualifier="numpy" value="ndarray%253A %255B%255B 0 1 2 ... 9997 9998 9999%255D%250A %255B10000 10001 10002 ... 19997 19998 19999%255D%250A %255B20000 20001 20002 ... 29997 29998 29999%255D%250A ...%250A %255B70000 70001 70002 ... 79997 79998 79999%255D%250A %255B80000 80001 80002 ... 89997 89998 89999%255D%250A %255B90000 90001 90002 ... 99997 99998 99999%255D%255D" isContainer="True" />',

'<var name="bigarray" type="ndarray" qualifier="numpy" value="ndarray%253A %255B%255B 0 1 2 ...%252C 9997 9998 9999%255D%250A %255B10000 10001 10002 ...%252C 19997 19998 19999%255D%250A %255B20000 20001 20002 ...%252C 29997 29998 29999%255D%250A ...%252C %250A %255B70000 70001 70002 ...%252C 79997 79998 79999%255D%250A %255B80000 80001 80002 ...%252C 89997 89998 89999%255D%250A %255B90000 90001 90002 ...%252C 99997 99998 99999%255D%255D" isContainer="True" />'

),

# Any of the ones below will do.

(

'<var name="hugearray" type="ndarray" qualifier="numpy" value="ndarray%253A %255B 0 1 2 ... 9999997 9999998 9999999%255D" isContainer="True" />',

'<var name="hugearray" type="ndarray" qualifier="numpy" value="ndarray%253A %255B 0 1 2 ...%252C 9999997 9999998 9999999%255D" isContainer="True" />'

)

))

# For each variable, check each of the resolved (meta data) attributes...

self.write_get_variable(hit.thread_id, hit.frame_id, 'smallarray')

self.wait_for_multiple_vars((

'<var name="min" type="complex128"',

'<var name="max" type="complex128"',

'<var name="shape" type="tuple"',

'<var name="dtype" type="dtype"',

'<var name="size" type="int"',

))

# ...and check that the internals are resolved properly

self.write_get_variable(hit.thread_id, hit.frame_id, 'smallarray\t__internals__')

self.wait_for_var('<var name="%27size%27')

self.write_get_variable(hit.thread_id, hit.frame_id, 'bigarray')

# isContainer could be true on some numpy versions, so, we only check for the var begin.

self.wait_for_multiple_vars((

[

'<var name="min" type="int64" qualifier="numpy" value="int64%253A 0"',

'<var name="min" type="int64" qualifier="numpy" value="int64%3A 0"',

'<var name="size" type="int" qualifier="{0}" value="int%3A 100000"'.format(builtin_qualifier),

],

[

'<var name="max" type="int64" qualifier="numpy" value="int64%253A 99999"',

'<var name="max" type="int32" qualifier="numpy" value="int32%253A 99999"',

'<var name="max" type="int64" qualifier="numpy" value="int64%3A 99999"',

'<var name="max" type="int32" qualifier="numpy" value="int32%253A 99999"',

],

'<var name="shape" type="tuple"',

'<var name="dtype" type="dtype"',

'<var name="size" type="int"'

))

self.write_get_variable(hit.thread_id, hit.frame_id, 'bigarray\t__internals__')

self.wait_for_var('<var name="%27size%27')

# this one is different because it crosses the magic threshold where we don't calculate

# the min/max

self.write_get_variable(hit.thread_id, hit.frame_id, 'hugearray')

self.wait_for_var((

[

'<var name="min" type="str" qualifier={0} value="str%253A ndarray too big%252C calculating min would slow down debugging" />'.format(builtin_qualifier),

'<var name="min" type="str" qualifier={0} value="str%3A ndarray too big%252C calculating min would slow down debugging" />'.format(builtin_qualifier),

'<var name="min" type="str" qualifier="{0}" value="str%253A ndarray too big%252C calculating min would slow down debugging" />'.format(builtin_qualifier),

'<var name="min" type="str" qualifier="{0}" value="str%3A ndarray too big%252C calculating min would slow down debugging" />'.format(builtin_qualifier),

],

[

'<var name="max" type="str" qualifier={0} value="str%253A ndarray too big%252C calculating max would slow down debugging" />'.format(builtin_qualifier),

'<var name="max" type="str" qualifier={0} value="str%3A ndarray too big%252C calculating max would slow down debugging" />'.format(builtin_qualifier),

'<var name="max" type="str" qualifier="{0}" value="str%253A ndarray too big%252C calculating max would slow down debugging" />'.format(builtin_qualifier),

'<var name="max" type="str" qualifier="{0}" value="str%3A ndarray too big%252C calculating max would slow down debugging" />'.format(builtin_qualifier),

],

'<var name="shape" type="tuple"',

'<var name="dtype" type="dtype"',

'<var name="size" type="int"',

))

self.write_get_variable(hit.thread_id, hit.frame_id, 'hugearray\t__internals__')

self.wait_for_var('<var name="%27size%27')

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase15 - [Test Case]: Custom Commands

#======================================================================================================================

class WriterThreadCase15(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case15.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(22, 'main')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit(REASON_STOP_ON_BREAKPOINT)

# Access some variable

self.write_custom_operation("%s\t%s\tEXPRESSION\tcarObj.color" % (hit.thread_id, hit.frame_id), "EXEC", "f=lambda x: 'val=%s' % x", "f")

self.wait_for_custom_operation('val=Black')

assert 7 == self._sequence, 'Expected 7. Had: %s' % self._sequence

self.write_custom_operation("%s\t%s\tEXPRESSION\tcarObj.color" % (hit.thread_id, hit.frame_id), "EXECFILE", debugger_unittest._get_debugger_test_file('_debugger_case15_execfile.py'), "f")

self.wait_for_custom_operation('val=Black')

assert 9 == self._sequence, 'Expected 9. Had: %s' % self._sequence

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase14 - [Test Case]: Interactive Debug Console

#======================================================================================================================

class WriterThreadCase14(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case14.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(22, 'main')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111')

assert hit.thread_id, '%s not valid.' % hit.thread_id

assert hit.frame_id, '%s not valid.' % hit.frame_id

# Access some variable

self.write_debug_console_expression("%s\t%s\tEVALUATE\tcarObj.color" % (hit.thread_id, hit.frame_id))

self.wait_for_var(['<more>False</more>', '%27Black%27'])

assert 7 == self._sequence, 'Expected 9. Had: %s' % self._sequence

# Change some variable

self.write_debug_console_expression("%s\t%s\tEVALUATE\tcarObj.color='Red'" % (hit.thread_id, hit.frame_id))

self.write_debug_console_expression("%s\t%s\tEVALUATE\tcarObj.color" % (hit.thread_id, hit.frame_id))

self.wait_for_var(['<more>False</more>', '%27Red%27'])

assert 11 == self._sequence, 'Expected 13. Had: %s' % self._sequence

# Iterate some loop

self.write_debug_console_expression("%s\t%s\tEVALUATE\tfor i in range(3):" % (hit.thread_id, hit.frame_id))

self.wait_for_var(['<xml><more>True</more></xml>'])

self.write_debug_console_expression("%s\t%s\tEVALUATE\t print(i)" % (hit.thread_id, hit.frame_id))

self.wait_for_var(['<xml><more>True</more></xml>'])

self.write_debug_console_expression("%s\t%s\tEVALUATE\t" % (hit.thread_id, hit.frame_id))

self.wait_for_var(

[

'<xml><more>False</more><output message="0"></output><output message="1"></output><output message="2"></output></xml>' ]

)

assert 17 == self._sequence, 'Expected 19. Had: %s' % self._sequence

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase13

#======================================================================================================================

class WriterThreadCase13(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case13.py')

def _ignore_stderr_line(self, line):

if debugger_unittest.AbstractWriterThread._ignore_stderr_line(self, line):

return True

if IS_JYTHON:

for expected in (

"RuntimeWarning: Parent module '_pydevd_bundle' not found while handling absolute import",

"import __builtin__"):

if expected in line:

return True

return False

def run(self):

self.start_socket()

self.write_add_breakpoint(35, 'main')

self.write("%s\t%s\t%s" % (CMD_SET_PROPERTY_TRACE, self.next_seq(), "true;false;false;true"))

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111')

self.write_get_frame(hit.thread_id, hit.frame_id)

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107', line=25)

# Should go inside setter method

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107')

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107', line=21)

# Should go inside getter method

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107')

# Disable property tracing

self.write("%s\t%s\t%s" % (CMD_SET_PROPERTY_TRACE, self.next_seq(), "true;true;true;true"))

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107', line=39)

# Should Skip step into properties setter

# Enable property tracing

self.write("%s\t%s\t%s" % (CMD_SET_PROPERTY_TRACE, self.next_seq(), "true;false;false;true"))

self.write_step_in(hit.thread_id)

hit = self.wait_for_breakpoint_hit('107', line=8)

# Should go inside getter method

self.write_run_thread(hit.thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase12

#======================================================================================================================

class WriterThreadCase12(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case_simple_calls.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(2, '') # Should not be hit: setting empty function (not None) should only hit global.

self.write_add_breakpoint(6, 'Method1a')

self.write_add_breakpoint(11, 'Method2')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111', line=11)

self.write_step_return(hit.thread_id)

hit = self.wait_for_breakpoint_hit('111', line=6) # not a return (it stopped in the other breakpoint)

self.write_run_thread(hit.thread_id)

assert 13 == self._sequence, 'Expected 13. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase11

#======================================================================================================================

class WriterThreadCase11(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case_simple_calls.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(2, 'Method1')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111', line=2)

self.write_step_over(hit.thread_id)

hit = self.wait_for_breakpoint_hit('108', line=3)

self.write_step_over(hit.thread_id)

hit = self.wait_for_breakpoint_hit('108', line=11)

self.write_step_over(hit.thread_id)

hit = self.wait_for_breakpoint_hit('108', line=12)

self.write_run_thread(hit.thread_id)

assert 13 == self._sequence, 'Expected 13. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase10

#======================================================================================================================

class WriterThreadCase10(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case_simple_calls.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(2, 'None') # None or Method should make hit.

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111')

self.write_step_return(hit.thread_id)

hit = self.wait_for_breakpoint_hit('109', line=11)

self.write_step_over(hit.thread_id)

hit = self.wait_for_breakpoint_hit('108', line=12)

self.write_run_thread(hit.thread_id)

assert 11 == self._sequence, 'Expected 11. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase9

#======================================================================================================================

class WriterThreadCase9(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case89.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(10, 'Method3')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111')

# Note: no active exception (should not give an error and should return no

# exception details as there's no exception).

self.write_get_current_exception(hit.thread_id)

msg = self.wait_for_message(accept_message=lambda msg:msg.strip().startswith(str(CMD_GET_EXCEPTION_DETAILS)))

assert msg.thread['id'] == hit.thread_id

assert not hasattr(msg.thread, 'frames') # No frames should be found.

self.write_step_over(hit.thread_id)

hit = self.wait_for_breakpoint_hit('108', line=11)

self.write_step_over(hit.thread_id)

hit = self.wait_for_breakpoint_hit('108', line=12)

self.write_run_thread(hit.thread_id)

assert 13 == self._sequence, 'Expected 13. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase8

#======================================================================================================================

class WriterThreadCase8(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case89.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(10, 'Method3')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111')

self.write_step_return(hit.thread_id)

hit = self.wait_for_breakpoint_hit('109', line=15)

self.write_run_thread(hit.thread_id)

assert 9 == self._sequence, 'Expected 9. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase7

#======================================================================================================================

class WriterThreadCase7(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case7.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(2, 'Call')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit('111')

self.write_get_frame(hit.thread_id, hit.frame_id)

self.wait_for_vars('<xml></xml>') # no vars at this point

self.write_step_over(hit.thread_id)

self.wait_for_breakpoint_hit('108')

self.write_get_frame(hit.thread_id, hit.frame_id)

self.wait_for_vars([

[

'<xml><var name="variable_for_test_1" type="int" qualifier="{0}" value="int%253A 10" />%0A</xml>'.format(builtin_qualifier),

'<var name="variable_for_test_1" type="int" value="int', # jython

]

])

self.write_step_over(hit.thread_id)

self.wait_for_breakpoint_hit('108')

self.write_get_frame(hit.thread_id, hit.frame_id)

self.wait_for_vars([

[

'<xml><var name="variable_for_test_1" type="int" qualifier="{0}" value="int%253A 10" />%0A<var name="variable_for_test_2" type="int" qualifier="{0}" value="int%253A 20" />%0A</xml>'.format(builtin_qualifier),

'<var name="variable_for_test_1" type="int" value="int%253A 10" />%0A<var name="variable_for_test_2" type="int" value="int%253A 20" />%0A', # jython

]

])

self.write_run_thread(hit.thread_id)

assert 17 == self._sequence, 'Expected 17. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase6

#=======================================================================================================================

class WriterThreadCase6(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case56.py')

def run(self):

self.start_socket()

self.write_add_breakpoint(2, 'Call2')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit()

thread_id = hit.thread_id

frame_id = hit.frame_id

self.write_get_frame(thread_id, frame_id)

self.write_step_return(thread_id)

hit = self.wait_for_breakpoint_hit('109')

thread_id = hit.thread_id

frame_id = hit.frame_id

line = hit.line

assert line == 8, 'Expecting it to go to line 8. Went to: %s' % line

self.write_step_in(thread_id)

hit = self.wait_for_breakpoint_hit('107')

thread_id = hit.thread_id

frame_id = hit.frame_id

line = hit.line

# goes to line 4 in jython (function declaration line)

assert line in (4, 5), 'Expecting it to go to line 4 or 5. Went to: %s' % line

self.write_run_thread(thread_id)

assert 13 == self._sequence, 'Expected 15. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase5

#=======================================================================================================================

class WriterThreadCase5(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case56.py')

def run(self):

self.start_socket()

breakpoint_id = self.write_add_breakpoint(2, 'Call2')

self.write_make_initial_run()

hit = self.wait_for_breakpoint_hit()

thread_id = hit.thread_id

frame_id = hit.frame_id

self.write_get_frame(thread_id, frame_id)

self.write_remove_breakpoint(breakpoint_id)

self.write_step_return(thread_id)

hit = self.wait_for_breakpoint_hit('109')

thread_id = hit.thread_id

frame_id = hit.frame_id

line = hit.line

assert line == 8, 'Expecting it to go to line 8. Went to: %s' % line

self.write_step_in(thread_id)

hit = self.wait_for_breakpoint_hit('107')

thread_id = hit.thread_id

frame_id = hit.frame_id

line = hit.line

# goes to line 4 in jython (function declaration line)

assert line in (4, 5), 'Expecting it to go to line 4 or 5. Went to: %s' % line

self.write_run_thread(thread_id)

assert 15 == self._sequence, 'Expected 15. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase4

#=======================================================================================================================

class WriterThreadCase4(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case4.py')

def run(self):

self.start_socket()

self.write_make_initial_run()

thread_id = self.wait_for_new_thread()

self.write_suspend_thread(thread_id)

hit = self.wait_for_breakpoint_hit(REASON_THREAD_SUSPEND)

assert hit.thread_id == thread_id

self.write_run_thread(thread_id)

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCase3

#=======================================================================================================================

class WriterThreadCase3(debugger_unittest.AbstractWriterThread):

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case3.py')

def run(self):

self.start_socket()

self.write_make_initial_run()

time.sleep(.5)

breakpoint_id = self.write_add_breakpoint(4, '')

self.write_add_breakpoint(5, 'FuncNotAvailable') # Check that it doesn't get hit in the global when a function is available

hit = self.wait_for_breakpoint_hit()

thread_id = hit.thread_id

frame_id = hit.frame_id

self.write_get_frame(thread_id, frame_id)

self.write_run_thread(thread_id)

hit = self.wait_for_breakpoint_hit()

thread_id = hit.thread_id

frame_id = hit.frame_id

self.write_get_frame(thread_id, frame_id)

self.write_remove_breakpoint(breakpoint_id)

self.write_run_thread(thread_id)

assert 17 == self._sequence, 'Expected 17. Had: %s' % self._sequence

self.finished_ok = True

#=======================================================================================================================

# WriterThreadCaseUnhandledExceptionsBasic

#=======================================================================================================================

class WriterThreadCaseUnhandledExceptionsBasic(debugger_unittest.AbstractWriterThread):

# Note: expecting unhandled exceptions to be printed to stdout.

TEST_FILE = debugger_unittest._get_debugger_test_file('_debugger_case_unhandled_exceptions.py')

@overrides(debugger_unittest.AbstractWriterThread.check_test_suceeded_msg)

def check_test_suceeded_msg(self, stdout, stderr):

return 'TEST SUCEEDED' in ''.join(stdout) and 'TEST SUCEEDED' in ''.join(stderr)

@overrides(debugger_unittest.AbstractWriterThread.additional_output_checks)

def additional_output_checks(self, stdout, stderr):

if 'raise Exception' not in stderr:

raise AssertionError('Expected test to have an unhandled exception.\nstdout:\n%s\n\nstderr:\n%s' % (

stdout, stderr))

# Don't call super (we have an unhandled exception in the stack trace).

def run(self):

self.start_socket()

self.write_add_exception_breakpoint_with_policy('Exception', "0", "1", "0")

self.write_make_initial_run()

def check(hit, exc_type, exc_desc):

self.write_get_current_exception(hit.thread_id)

msg = self.wait_for_message(accept_message=lambda msg:exc_type in msg and 'exc_type="' in msg and 'exc_desc="' in msg, unquote_msg=False)

assert unquote(msg.thread['exc_desc']) == exc_desc

assert unquote(msg.thread['exc_type']) in (

"&lt;type 'exceptions.%s'&gt;" % (exc_type,), # py2

"&lt;class '%s'&gt;" % (exc_type,) # py3

)

if len(msg.thread.frame) == 0:

assert unquote(unquote(msg.thread.frame['file'])).endswith('_debugger_case_unhandled_exceptions.py')

else:

assert unquote(unquote(msg.thread.frame[0]['file'])).endswith('_debugger_case_unhandled_exceptions.py')

self.write_run_thread(hit.thread_id)

# Will stop in 2 background threads

hit0 = self.wait_for_breakpoint_hit(REASON_UNCAUGHT_EXCEPTION)

thread_id1 = hit0.thread_id

hit1 = self.wait_for_breakpoint_hit(REASON_UNCAUGHT_EXCEPTION)

thread_id2 = hit1.thread_id

if hit0.name == 'thread_func2':

check(hit0, 'ValueError', 'in thread 2')

check(hit1, 'Exception', 'in thread 1')

else:

check(hit0, 'Exception', 'in thread 1')

check(hit1, 'ValueError', 'in thread 2')

self.write_run_thread(thread_id1)

self.write_run_thread(thread_id2)

# Will stop in main thread

hit = self.wait_for_breakpoint_hit(REASON_UNCAUGHT_EXCEPTION)

assert hit.name == '<module>'

thread_id3 = hit.thread_id

# Requesting the stack in an unhandled exception should provide the stack of the exception,

# not the current location of the program.

self.write_get_thread_stack(thread_id3)

msg = self.wait_for_message(lambda msg:msg.startswith('%s\t' % (CMD_GET_THREAD_STACK,)))

assert len(msg.thread.frame) == 0 # In main thread (must have no back frames).