<commit_before>"""
Test lldb-vscode setBreakpoints request
"""

from __future__ import print_function

import unittest2
import vscode
from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *
from lldbsuite.test import lldbutil
import lldbvscode_testcase
import os
import shutil
import subprocess
import tempfile
import threading
import time


def spawn_and_wait(program, delay):
    if delay:
        time.sleep(delay)
    process = subprocess.Popen([program],
                               stdin=subprocess.PIPE,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
    process.wait()


class TestVSCode_attach(lldbvscode_testcase.VSCodeTestCaseBase):

    mydir = TestBase.compute_mydir(__file__)

    def set_and_hit_breakpoint(self, continueToExit=True):
        source = 'main.c'
        breakpoint1_line = line_number(source, '// breakpoint 1')
        lines = [breakpoint1_line]
        # Set breakoint in the thread function so we can step the threads
        breakpoint_ids = self.set_source_breakpoints(source, lines)
        self.assertEqual(len(breakpoint_ids), len(lines),
                         "expect correct number of breakpoints")
        self.continue_to_breakpoints(breakpoint_ids)
        if continueToExit:
            self.continue_to_exit()


    @skipIfWindows
    @no_debug_info_test
    def test_by_pid(self):
        '''
            Tests attaching to a process by process ID.
        '''
        self.build_and_create_debug_adaptor()
        program = self.getBuildArtifact("a.out")
        self.process = subprocess.Popen([program],
                                        stdin=subprocess.PIPE,
                                        stdout=subprocess.PIPE,
                                        stderr=subprocess.PIPE)
        self.attach(pid=self.process.pid)
        self.set_and_hit_breakpoint(continueToExit=True)

    @skipIfWindows
    @no_debug_info_test
    def test_by_name(self):
        '''
            Tests attaching to a process by process name.
        '''
        self.build_and_create_debug_adaptor()
        orig_program = self.getBuildArtifact("a.out")
        # Since we are going to attach by process name, we need a unique
        # process name that has minimal chance to match a process that is
        # already running. To do this we use tempfile.mktemp() to give us a
        # full path to a location where we can copy our executable. We then
        # run this copy to ensure we don't get the error "more that one
        # process matches 'a.out'".
        program = tempfile.mktemp()
        shutil.copyfile(orig_program, program)
        shutil.copymode(orig_program, program)

        def cleanup():
            if os.path.exists(program):
                os.unlink(program)
        # Execute the cleanup function during test case tear down.
        self.addTearDownHook(cleanup)

        self.process = subprocess.Popen([program],
                                        stdin=subprocess.PIPE,
                                        stdout=subprocess.PIPE,
                                        stderr=subprocess.PIPE)
        # Wait for a bit to ensure the process is launched
        time.sleep(1)
        self.attach(program=program)
        self.set_and_hit_breakpoint(continueToExit=True)

    @skipUnlessDarwin
    @no_debug_info_test
    def test_by_name_waitFor(self):
        '''
            Tests attaching to a process by process name and waiting for the
            next instance of a process to be launched, ingoring all current
            ones.
        '''
        self.build_and_create_debug_adaptor()
        program = self.getBuildArtifact("a.out")
        self.spawn_thread = threading.Thread(target=spawn_and_wait,
                                             args=(program, 1.0,))
        self.spawn_thread.start()
        self.attach(program=program, waitFor=True)
        self.set_and_hit_breakpoint(continueToExit=True)

    @skipIfWindows
    @no_debug_info_test
    def test_commands(self):
        '''<commit_msg>Skip the lldb-vscode attach with waitFor test for now, some build bots are having trouble.<commit_after>"""
Test lldb-vscode setBreakpoints request
"""

from __future__ import print_function

import unittest2
import vscode
from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *
from lldbsuite.test import lldbutil
import lldbvscode_testcase
import os
import shutil
import subprocess
import tempfile
import threading
import time


def spawn_and_wait(program, delay):
    if delay:
        time.sleep(delay)
    process = subprocess.Popen([program],
                               stdin=subprocess.PIPE,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
    process.wait()


class TestVSCode_attach(lldbvscode_testcase.VSCodeTestCaseBase):

    mydir = TestBase.compute_mydir(__file__)

    def set_and_hit_breakpoint(self, continueToExit=True):
        source = 'main.c'
        breakpoint1_line = line_number(source, '// breakpoint 1')
        lines = [breakpoint1_line]
        # Set breakoint in the thread function so we can step the threads
        breakpoint_ids = self.set_source_breakpoints(source, lines)
        self.assertEqual(len(breakpoint_ids), len(lines),
                         "expect correct number of breakpoints")
        self.continue_to_breakpoints(breakpoint_ids)
        if continueToExit:
            self.continue_to_exit()


    @skipIfWindows
    @no_debug_info_test
    def test_by_pid(self):
        '''
            Tests attaching to a process by process ID.
        '''
        self.build_and_create_debug_adaptor()
        program = self.getBuildArtifact("a.out")
        self.process = subprocess.Popen([program],
                                        stdin=subprocess.PIPE,
                                        stdout=subprocess.PIPE,
                                        stderr=subprocess.PIPE)
        self.attach(pid=self.process.pid)
        self.set_and_hit_breakpoint(continueToExit=True)

    @skipIfWindows
    @no_debug_info_test
    def test_by_name(self):
        '''
            Tests attaching to a process by process name.
        '''
        self.build_and_create_debug_adaptor()
        orig_program = self.getBuildArtifact("a.out")
        # Since we are going to attach by process name, we need a unique
        # process name that has minimal chance to match a process that is
        # already running. To do this we use tempfile.mktemp() to give us a
        # full path to a location where we can copy our executable. We then
        # run this copy to ensure we don't get the error "more that one
        # process matches 'a.out'".
        program = tempfile.mktemp()
        shutil.copyfile(orig_program, program)
        shutil.copymode(orig_program, program)

        def cleanup():
            if os.path.exists(program):
                os.unlink(program)
        # Execute the cleanup function during test case tear down.
        self.addTearDownHook(cleanup)

        self.process = subprocess.Popen([program],
                                        stdin=subprocess.PIPE,
                                        stdout=subprocess.PIPE,
                                        stderr=subprocess.PIPE)
        # Wait for a bit to ensure the process is launched
        time.sleep(1)
        self.attach(program=program)
        self.set_and_hit_breakpoint(continueToExit=True)

    @skipUnlessDarwin
    @skipIfDarwin # Skip this test for now until we can figure out why tings aren't working on build bots
    @no_debug_info_test
    def test_by_name_waitFor(self):
        '''
            Tests attaching to a process by process name and waiting for the
            next instance of a process to be launched, ingoring all current
            ones.
        '''
        self.build_and_create_debug_adaptor()
        program = self.getBuildArtifact("a.out")
        self.spawn_thread = threading.Thread(target=spawn_and_wait,
                                             args=(program, 1.0,))
        self.spawn_thread.start()
        self.attach(program=program, waitFor=True)
        self.set_and_hit_breakpoint(continueToExit=True)

    @skipIfWindows
    @no_debug_info_test
    def test_commands(self):
        '''
{'fraction_non_alphanumeric': 0.07193213048153901, 'fraction_numerical': 0.0016728402437567213, 'mean_word_length': 2.0977054034048854, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3112274', 'n_tokens_mistral': 2334, 'n_tokens_neox': 2273, 'n_words': 582}