debugpy/tests/func/test_breakpoints.py

# -*- coding: utf-8 -*-
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See LICENSE in the project root
# for license information.

from __future__ import print_function, with_statement, absolute_import

import os.path
import pytest
import sys
import re

from tests.helpers.pathutils import get_test_root
from tests.helpers.session import DebugSession
from tests.helpers.timeline import Event
from tests.helpers.pattern import ANY, Path


BP_TEST_ROOT = get_test_root('bp')


def test_path_with_ampersand(run_as, start_method):
    bp_line = 4
    testfile = os.path.join(BP_TEST_ROOT, 'a&b', 'test.py')

    with DebugSession() as session:
        session.initialize(
            target=(run_as, testfile),
            start_method=start_method,
            ignore_unobserved=[Event('continued')],
        )
        session.set_breakpoints(testfile, [bp_line])
        session.start_debugging()
        hit = session.wait_for_thread_stopped('breakpoint')
        frames = hit.stacktrace.body['stackFrames']
        assert frames[0]['source']['path'] == Path(testfile)

        session.send_request('continue').wait_for_response(freeze=False)
        session.wait_for_exit()


@pytest.mark.skipif(sys.version_info < (3, 0), reason='Paths are not Unicode in Python 2.7')
def test_path_with_unicode(run_as, start_method):
    bp_line = 6
    testfile = os.path.join(BP_TEST_ROOT, u'ನನ್ನ_ಸ್ಕ್ರಿಪ್ಟ್.py')

    with DebugSession() as session:
        session.initialize(
            target=(run_as, testfile),
            start_method=start_method,
            ignore_unobserved=[Event('continued')],
        )
        session.set_breakpoints(testfile, [bp_line])
        session.start_debugging()
        hit = session.wait_for_thread_stopped('breakpoint')
        frames = hit.stacktrace.body['stackFrames']
        assert frames[0]['source']['path'] == Path(testfile)
        assert u'ಏನಾದರೂ_ಮಾಡು' == frames[0]['name']

        session.send_request('continue').wait_for_response(freeze=False)
        session.wait_for_exit()


@pytest.mark.parametrize('condition_key', [
    'condition_var',
    'hitCondition_#',
    'hitCondition_eq',
    'hitCondition_gt',
    'hitCondition_ge',
    'hitCondition_lt',
    'hitCondition_le',
    'hitCondition_mod',
])
def test_conditional_breakpoint(pyfile, run_as, start_method, condition_key):
    @pyfile
    def code_to_debug():
        from dbgimporter import import_and_enable_debugger
        import_and_enable_debugger()
        for i in range(0, 10):
            print(i)

    expected = {
        'condition_var': ('condition', 'i==5', '5', 1),
        'hitCondition_#': ('hitCondition', '5', '4', 1),
        'hitCondition_eq': ('hitCondition', '==5', '4', 1),
        'hitCondition_gt': ('hitCondition', '>5', '5', 5),
        'hitCondition_ge': ('hitCondition', '>=5', '4', 6),
        'hitCondition_lt': ('hitCondition', '<5', '0', 4),
        'hitCondition_le': ('hitCondition', '<=5', '0', 5),
        'hitCondition_mod': ('hitCondition', '%3', '2', 3),
    }
    condition_type, condition, value, hits = expected[condition_key]

    bp_line = 4
    with DebugSession() as session:
        session.initialize(
            target=(run_as, code_to_debug),
            start_method=start_method,
            ignore_unobserved=[Event('continued')],
        )
        session.send_request('setBreakpoints', arguments={
            'source': {'path': code_to_debug},
            'breakpoints': [{'line': bp_line, condition_type: condition}],
        }).wait_for_response()
        session.start_debugging()
        hit = session.wait_for_thread_stopped()
        frames = hit.stacktrace.body['stackFrames']
        assert bp_line == frames[0]['line']

        resp_scopes = session.send_request('scopes', arguments={
            'frameId': hit.frame_id
        }).wait_for_response()
        scopes = resp_scopes.body['scopes']
        assert len(scopes) > 0

        resp_variables = session.send_request('variables', arguments={
            'variablesReference': scopes[0]['variablesReference']
        }).wait_for_response()
        variables = list(v for v in resp_variables.body['variables']
                         if v['name'] == 'i')
        assert variables == [
            ANY.dict_with({'name': 'i', 'type': 'int', 'value': value, 'evaluateName': 'i'})
        ]

        session.send_request('continue').wait_for_response(freeze=False)
        for i in range(1, hits):
            session.wait_for_thread_stopped()
            session.send_request('continue').wait_for_response(freeze=False)
        session.wait_for_exit()


def test_crossfile_breakpoint(pyfile, run_as, start_method):
    @pyfile
    def script1():
        from dbgimporter import import_and_enable_debugger  # noqa
        def do_something():
            print('do something')

    @pyfile
    def script2():
        from dbgimporter import import_and_enable_debugger
        import_and_enable_debugger()
        import script1
        script1.do_something()
        print('Done')

    bp_script1_line = 3
    bp_script2_line = 4
    with DebugSession() as session:
        session.initialize(
            target=(run_as, script2),
            start_method=start_method,
            ignore_unobserved=[Event('continued')],
        )
        session.set_breakpoints(script1, lines=[bp_script1_line])
        session.set_breakpoints(script2, lines=[bp_script2_line])
        session.start_debugging()

        hit = session.wait_for_thread_stopped()
        frames = hit.stacktrace.body['stackFrames']
        assert bp_script2_line == frames[0]['line']
        assert frames[0]['source']['path'] == Path(script2)

        session.send_request('continue').wait_for_response(freeze=False)
        hit = session.wait_for_thread_stopped()
        frames = hit.stacktrace.body['stackFrames']
        assert bp_script1_line == frames[0]['line']
        assert frames[0]['source']['path'] == Path(script1)

        session.send_request('continue').wait_for_response(freeze=False)
        session.wait_for_exit()


@pytest.mark.parametrize('error_name', [
    'NameError',
    'OtherError',
])
def test_error_in_condition(pyfile, run_as, start_method, error_name):
    @pyfile
    def code_to_debug():
        from dbgimporter import import_and_enable_debugger
        import_and_enable_debugger()
        def do_something_bad():
            raise ArithmeticError()
        for i in range(1, 10):
            pass

    # NOTE: NameError in condition, is a special case. Pydevd is configured to skip
    # traceback for name errors. See https://github.com/Microsoft/ptvsd/issues/853
    # for more details. For all other errors we should be printing traceback.
    condition = {
        'NameError': ('x==5'),  # 'x' does not exist in the debuggee
        'OtherError': ('do_something_bad()==5')  # throws some error
    }

    bp_line = 5
    with DebugSession() as session:
        session.initialize(
            target=(run_as, code_to_debug),
            start_method=start_method,
            ignore_unobserved=[Event('continued')],
        )
        session.send_request('setBreakpoints', arguments={
            'source': {'path': code_to_debug},
            'breakpoints': [{
                'line': bp_line,
                'condition': condition[error_name],
            }],
        }).wait_for_response()
        session.start_debugging()

        session.wait_for_exit()
        assert session.get_stdout_as_string() == b''
        if error_name == 'NameError':
            assert session.get_stderr_as_string().find(b'NameError') == -1
        else:
            assert session.get_stderr_as_string().find(b'ArithmeticError') > 0


@pytest.mark.skip(reason='bug #799')
def test_log_point(pyfile, run_as, start_method):
    @pyfile
    def code_to_debug():
        from dbgimporter import import_and_enable_debugger
        import_and_enable_debugger()
        a = 10
        for i in range(1, a):
            print('value: %d' % i)

    bp_line = 5
    with DebugSession() as session:
        session.initialize(
            target=(run_as, code_to_debug),
            start_method=start_method,
            ignore_unobserved=[Event('continued')],
        )
        session.send_request('setBreakpoints', arguments={
            'source': {'path': code_to_debug},
            'breakpoints': [{
                'line': bp_line,
                'logMessage': 'log: {a + i}'
            }],
        }).wait_for_response()
        session.start_debugging()

        session.wait_for_exit()
        assert session.get_stderr_as_string() == b''

        output = session.all_occurrences_of(Event('output', ANY.dict_with({'category': 'stdout'})))
        output_str = ''.join(o.body['output'] for o in output)
        logged = sorted(int(i) for i in re.findall(r"log:\s([0-9]*)", output_str))
        values = sorted(int(i) for i in re.findall(r"value:\s([0-9]*)", output_str))

        # NOTE: Due to https://github.com/Microsoft/ptvsd/issues/1028 we may not get
        # all output events. Once that is fixed we should check for the exact output
        # and log value
        assert len(logged) > 0
        assert len(values) > 0
        # assert logged == list(range(11, 20))
        # assert values == list(range(1, 10))


@pytest.mark.skip(reason='bug #799')
def test_condition_with_log_point(pyfile, run_as, start_method):
    @pyfile
    def code_to_debug():
        from dbgimporter import import_and_enable_debugger
        import_and_enable_debugger()
        a = 10
        for i in range(1, a):
            print('value: %d' % i)

    bp_line = 5
    with DebugSession() as session:
        session.initialize(
            target=(run_as, code_to_debug),
            start_method=start_method,
            ignore_unobserved=[Event('continued')],
        )
        session.send_request('setBreakpoints', arguments={
            'source': {'path': code_to_debug},
            'breakpoints': [{
                'line': bp_line,
                'logMessage': 'log: {a + i}',
                'condition': 'i==5'
            }],
        }).wait_for_response()
        session.start_debugging()
        hit = session.wait_for_thread_stopped()
        frames = hit.stacktrace.body['stackFrames']
        assert bp_line == frames[0]['line']

        resp_scopes = session.send_request('scopes', arguments={
            'frameId': hit.frame_id
        }).wait_for_response()
        scopes = resp_scopes.body['scopes']
        assert len(scopes) > 0

        resp_variables = session.send_request('variables', arguments={
            'variablesReference': scopes[0]['variablesReference']
        }).wait_for_response()
        variables = list(
            v for v in resp_variables.body['variables']
            if v['name'] == 'i'
        )
        assert variables == [
            ANY.dict_with({'name': 'i', 'type': 'int', 'value': '5', 'evaluateName': 'i'})
        ]

        session.send_request('continue').wait_for_response(freeze=False)
        session.wait_for_exit()
        assert session.get_stderr_as_string() == b''

        output = session.all_occurrences_of(Event('output', ANY.dict_with({'category': 'stdout'})))
        output_str = ''.join(o.body['output'] for o in output)
        logged = sorted(int(i) for i in re.findall(r"log:\s([0-9]*)", output_str))
        values = sorted(int(i) for i in re.findall(r"value:\s([0-9]*)", output_str))

        assert 15 in logged
        assert 5 in values


@pytest.mark.skip(reason='Bug #1010')
def test_package_launch():
    bp_line = 2
    cwd = get_test_root('testpkgs')
    testfile = os.path.join(cwd, 'pkg1', '__main__.py')

    with DebugSession() as session:
        session.initialize(
            target=('module', 'pkg1'),
            start_method='launch',
            ignore_unobserved=[Event('continued')],
            cwd=cwd,
        )
        session.set_breakpoints(testfile, [bp_line])
        session.start_debugging()

        hit = session.wait_for_thread_stopped()
        frames = hit.stacktrace.body['stackFrames']
        assert bp_line == frames[0]['line']

        session.send_request('continue').wait_for_response(freeze=False)
        session.wait_for_exit()