<commit_before>import sys
import logging
import os.path
import shlex
import traceback
import lldb
from . import debugevents
from . import handles
from . import terminal
from . import PY2

log = logging.getLogger('debugsession')

class DebugSession:

    def __init__(self, event_loop, send_message):
        DebugSession.current = self
        self.event_loop = event_loop
        self.send_message = send_message
        self.var_refs = handles.Handles()
        self.breakpoints = dict() # { file : { line : SBBreakpoint } }
        self.fn_breakpoints = dict() # { name : SBBreakpoint }
        self.exc_breakpoints = []
        self.target = None
        self.process = None
        self.terminal = None
        self.launch_args = None

    def initialize_request(self, args):
        self.line_offset = 0 if args.get('linesStartAt1', True) else 1
        self.col_offset = 0 if args.get('columnsStartAt1', True) else 1
        self.debugger = lldb.SBDebugger.Create()
        log.info('LLDB version: %s', self.debugger.GetVersionString())
        self.debugger.SetAsync(True)
        self.event_listener = lldb.SBListener('DebugSession')
        listener_handler = debugevents.AsyncListener(self.event_listener,
                self.event_loop.make_dispatcher(self.handle_debugger_event))
        self.listener_handler_token = listener_handler.start()
        return { 'supportsConfigurationDoneRequest': True,
                 'supportsEvaluateForHovers': True,
                 'supportsFunctionBreakpoints': True,
                 'supportsConditionalBreakpoints': True }

    def launch_request(self, args):
        self.exec_commands(args.get('initCommands'))
        self.target = self.create_target(args)
        self.send_event('initialized', {})
        # defer actual launching till configurationDone request, so that
        # we can receive and set initial breakpoints before the target starts running
        self.do_launch = self.launch
        self.launch_args = args
        return AsyncResponse

    def launch(self, args):
        self.exec_commands(args.get('preRunCommands'))
        flags = 0
        # argumetns
        target_args = args.get('args', None)
        if target_args is not None:
            if isinstance(target_args, string_type):
                target_args = shlex.split(target_args)
            target_args = [str(arg) for arg in target_args]
        # environment
        env = args.get('env', None)
        envp = [str('%s=%s' % pair) for pair in os.environ.items()]
        if (env is not None): # Convert dict to a list of 'key=value' strings
            envp = envp + ([str('%s=%s' % pair) for pair in env.items()])
        # stdio
        stdio = self.get_stdio_config(args)
        # open a new terminal window if needed
        if '*' in stdio:
            if 'linux' in sys.platform:
                self.terminal = terminal.create()
                stdio = [self.terminal.tty if s == '*' else s for s in stdio]
            else:
                # OSX LLDB supports this natively.
                # On Windows LLDB always creates a new console window (even if stdio is redirected).
                flags |= lldb.eLaunchFlagLaunchInTTY | lldb.eLaunchFlagCloseTTYOnExit
                stdio = [None if s == '*' else s for s in stdio]
        # working directory
        work_dir = opt_str(args.get('cwd', None))
        stop_on_entry = args.get('stopOnEntry', False)
        # launch!
        error = lldb.SBError()
        self.process = self.target.Launch(self.event_listener,
            target_args, envp, stdio[0], stdio[1], stdio[2],
            work_dir, flags, stop_on_entry, error)
        if not error.Success():
            self.console_msg(error.GetCString())
            self.send_event('terminated', {})
            raise UserError('Process launch failed.')

        assert self.process.IsValid()
        self.process_launched = True

    def get_stdio_config(self, args):
        stdio = args.get('stdio', None)
        missing = () # None is a valid value here, so we need a new one to designate 'missing'
        if isinstance(stdio, dict): # Flatten it into a list
            stdio = [stdio.get('stdin', missing),
                     stdio.get('stdout', missing),
                     stdio.get('stderr', missing)]
        elif stdio is None or isinstance(stdio, string_type):
            stdio = [stdio] * 3
        elif isinstance(stdio, list):
            stdio.extend([missing] * (3-len(stdio))) # pad up to 3 items
        else:
            raise UserError('stdio must be either a string, a list or an object')
        # replace all missing's with the previous stream's value
        for i in range(0, len(stdio)):
            if stdio[i] == missing:
                stdio[i] = stdio[i-1] if i > 0 else None
        stdio = list(map(opt_str, stdio))
        return stdio

    def attach_request(self, args):
        self.exec_commands(args.get('initCommands'))
        self.target = self.create_target(args)
        self.send_event('initialized', {})
        self.do_launch = self.attach
        self.launch_args = args
        return AsyncResponse

    def attach(self, args):
        self.exec_commands(args.get('preRunCommands'))

        error = lldb.SBError()
        if args.get('pid', None) is not None:
            self.process = self.target.AttachToProcessWithID(self.event_listener, args['pid'], error)
        else:
            self.process = self.target.AttachToProcessWithName(self.event_listener, str(args['program']), False, error)

        if not error.Success():
            self.console_msg(error.GetCString())
            raise UserError('Failed to attach to process.')

        assert self.process.IsValid()
        self.process_launched = False

    def create_target(self, args):
        program = args['program']
        load_dependents = not args.get('noDebug', False)
        error = lldb.SBError()
        target = self.debugger.CreateTarget(str(program), lldb.LLDB_ARCH_DEFAULT, None, load_dependents, error)
        if not error.Success() and 'win' in sys.platform:
            # On Windows, try appending '.exe' extension, to make launch configs more uniform.
            program += '.exe'
            target = self.debugger.CreateTarget(str(program), lldb.LLDB_ARCH_DEFAULT, None, load_dependents, error)
            if error.Success():
                args['program'] = program
        if not error.Success():
            raise UserError('Could not initialize debug target: ' + error.GetCString())
        return target

    def exec_commands(self, commands):
        if commands is not None:
            interp = self.debugger.GetCommandInterpreter()
            result = lldb.SBCommandReturnObject()<commit_msg>Don't try ...+'.exe' on OSX.<commit_after>import sys
import logging
import os.path
import shlex
import traceback
import lldb
from . import debugevents
from . import handles
from . import terminal
from . import PY2

log = logging.getLogger('debugsession')

class DebugSession:

    def __init__(self, event_loop, send_message):
        DebugSession.current = self
        self.event_loop = event_loop
        self.send_message = send_message
        self.var_refs = handles.Handles()
        self.breakpoints = dict() # { file : { line : SBBreakpoint } }
        self.fn_breakpoints = dict() # { name : SBBreakpoint }
        self.exc_breakpoints = []
        self.target = None
        self.process = None
        self.terminal = None
        self.launch_args = None

    def initialize_request(self, args):
        self.line_offset = 0 if args.get('linesStartAt1', True) else 1
        self.col_offset = 0 if args.get('columnsStartAt1', True) else 1
        self.debugger = lldb.SBDebugger.Create()
        log.info('LLDB version: %s', self.debugger.GetVersionString())
        self.debugger.SetAsync(True)
        self.event_listener = lldb.SBListener('DebugSession')
        listener_handler = debugevents.AsyncListener(self.event_listener,
                self.event_loop.make_dispatcher(self.handle_debugger_event))
        self.listener_handler_token = listener_handler.start()
        return { 'supportsConfigurationDoneRequest': True,
                 'supportsEvaluateForHovers': True,
                 'supportsFunctionBreakpoints': True,
                 'supportsConditionalBreakpoints': True }

    def launch_request(self, args):
        self.exec_commands(args.get('initCommands'))
        self.target = self.create_target(args)
        self.send_event('initialized', {})
        # defer actual launching till configurationDone request, so that
        # we can receive and set initial breakpoints before the target starts running
        self.do_launch = self.launch
        self.launch_args = args
        return AsyncResponse

    def launch(self, args):
        self.exec_commands(args.get('preRunCommands'))
        flags = 0
        # argumetns
        target_args = args.get('args', None)
        if target_args is not None:
            if isinstance(target_args, string_type):
                target_args = shlex.split(target_args)
            target_args = [str(arg) for arg in target_args]
        # environment
        env = args.get('env', None)
        envp = [str('%s=%s' % pair) for pair in os.environ.items()]
        if (env is not None): # Convert dict to a list of 'key=value' strings
            envp = envp + ([str('%s=%s' % pair) for pair in env.items()])
        # stdio
        stdio = self.get_stdio_config(args)
        # open a new terminal window if needed
        if '*' in stdio:
            if 'linux' in sys.platform:
                self.terminal = terminal.create()
                stdio = [self.terminal.tty if s == '*' else s for s in stdio]
            else:
                # OSX LLDB supports this natively.
                # On Windows LLDB always creates a new console window (even if stdio is redirected).
                flags |= lldb.eLaunchFlagLaunchInTTY | lldb.eLaunchFlagCloseTTYOnExit
                stdio = [None if s == '*' else s for s in stdio]
        # working directory
        work_dir = opt_str(args.get('cwd', None))
        stop_on_entry = args.get('stopOnEntry', False)
        # launch!
        error = lldb.SBError()
        self.process = self.target.Launch(self.event_listener,
            target_args, envp, stdio[0], stdio[1], stdio[2],
            work_dir, flags, stop_on_entry, error)
        if not error.Success():
            self.console_msg(error.GetCString())
            self.send_event('terminated', {})
            raise UserError('Process launch failed.')

        assert self.process.IsValid()
        self.process_launched = True

    def get_stdio_config(self, args):
        stdio = args.get('stdio', None)
        missing = () # None is a valid value here, so we need a new one to designate 'missing'
        if isinstance(stdio, dict): # Flatten it into a list
            stdio = [stdio.get('stdin', missing),
                     stdio.get('stdout', missing),
                     stdio.get('stderr', missing)]
        elif stdio is None or isinstance(stdio, string_type):
            stdio = [stdio] * 3
        elif isinstance(stdio, list):
            stdio.extend([missing] * (3-len(stdio))) # pad up to 3 items
        else:
            raise UserError('stdio must be either a string, a list or an object')
        # replace all missing's with the previous stream's value
        for i in range(0, len(stdio)):
            if stdio[i] == missing:
                stdio[i] = stdio[i-1] if i > 0 else None
        stdio = list(map(opt_str, stdio))
        return stdio

    def attach_request(self, args):
        self.exec_commands(args.get('initCommands'))
        self.target = self.create_target(args)
        self.send_event('initialized', {})
        self.do_launch = self.attach
        self.launch_args = args
        return AsyncResponse

    def attach(self, args):
        self.exec_commands(args.get('preRunCommands'))

        error = lldb.SBError()
        if args.get('pid', None) is not None:
            self.process = self.target.AttachToProcessWithID(self.event_listener, args['pid'], error)
        else:
            self.process = self.target.AttachToProcessWithName(self.event_listener, str(args['program']), False, error)

        if not error.Success():
            self.console_msg(error.GetCString())
            raise UserError('Failed to attach to process.')

        assert self.process.IsValid()
        self.process_launched = False

    def create_target(self, args):
        program = args['program']
        load_dependents = not args.get('noDebug', False)
        error = lldb.SBError()
        target = self.debugger.CreateTarget(str(program), lldb.LLDB_ARCH_DEFAULT, None, load_dependents, error)
        if not error.Success() and 'win32' in sys.platform:
            # On Windows, try appending '.exe' extension, to make launch configs more uniform.
            program += '.exe'
            error2 = lldb.SBError()
            target = self.debugger.CreateTarget(str(program), lldb.LLDB_ARCH_DEFAULT, None, load_dependents, error2)
            if error2.Success():
                args['program'] = program
        if not error.Success():
            raise UserError('Could not initialize debug target: ' + error.GetCString())
        return target

    def exec_commands(self, commands):
        if commands is not None:
            interp = self.debugger.GetCommandInterpreter()
            result = lldb.SBCommandReturnObject()
{'fraction_non_alphanumeric': 0.10289651085265575, 'fraction_numerical': 0.0028891028965108527, 'mean_word_length': 2.330044400592008, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6238040', 'n_tokens_mistral': 3903, 'n_tokens_neox': 3773, 'n_words': 1015}