<commit_before>from contextlib import contextmanager
import faulthandler
import re
import signal
import subprocess
import sys
from test import support, script_helper
import tempfile
import unittest

try:
    from resource import setrlimit, RLIMIT_CORE, error as resource_error
except ImportError:
    prepare_subprocess = None
else:
    def prepare_subprocess():
        # don't create core file
        try:
            setrlimit(RLIMIT_CORE, (0, 0))
        except (ValueError, resource_error):
            pass

def expected_traceback(lineno1, lineno2, header, count=1):
    regex = header
    regex += r'  File "\<string\>", line %s in func\n' % lineno1
    regex += r'  File "\<string\>", line %s in \<module\>' % lineno2
    if count != 1:
        regex = (regex + '\n') * (count - 1) + regex
    return '^' + regex + '$'

@contextmanager
def temporary_filename():
    filename = tempfile.mktemp()
    try:
        yield filename
    finally:
        support.unlink(filename)

class FaultHandlerTests(unittest.TestCase):
    def get_output(self, code, expect_success, filename=None):
        """
        Run the specified code in Python (in a new child process) and read the
        output from the standard error or from a file (if filename is set).
        Return the output lines as a list.

        Strip the reference count from the standard error for Python debug
        build, and replace "Current thread 0x00007f8d8fbd9700" by "Current
        thread XXX".
        """
        options = {}
        if prepare_subprocess:
            options['preexec_fn'] = prepare_subprocess
        process = script_helper.spawn_python('-c', code, **options)
        stdout, stderr = process.communicate()
        exitcode = process.wait()
        if expect_success:
            self.assertEqual(exitcode, 0)
        else:
            self.assertNotEqual(exitcode, 0)
        if filename:
            with open(filename, "rb") as fp:
                output = fp.read()
        else:
            output = support.strip_python_stderr(stdout)
        output = output.decode('ascii', 'backslashreplace')
        output = re.sub('Current thread 0x[0-9a-f]+',
                        'Current thread XXX',
                        output)
        return output.splitlines()

    def check_fatal_error(self, code, line_number, name_regex,
                               filename=None, all_threads=False):
        """
        Check that the fault handler for fatal errors is enabled and check the
        traceback from the child process output.

        Raise an error if the output doesn't match the expected format.
        """
        if all_threads:
            header = 'Current thread XXX'
        else:
            header = 'Traceback (most recent call first)'
        regex = """
^Fatal Python error: {name}

{header}:
  File "<string>", line {lineno} in <module>$
""".strip()
        regex = regex.format(
            lineno=line_number,
            name=name_regex,
            header=re.escape(header))
        output = self.get_output(code, False, filename)
        output = '\n'.join(output)
        self.assertRegex(output, regex)

    def test_read_null(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._read_null()
""".strip(),
            3,
            '(?:Segmentation fault|Bus error)')

    def test_sigsegv(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigsegv()
""".strip(),
            3,
            'Segmentation fault')

    @unittest.skipIf(sys.platform == 'win32',
                     "SIGFPE cannot be caught on Windows")
    def test_sigfpe(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigfpe()
""".strip(),
            3,
            'Floating point exception')

    @unittest.skipIf(not hasattr(faulthandler, '_sigbus'),
                     "need faulthandler._sigbus()")
    def test_sigbus(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigbus()
""".strip(),
            3,
            'Bus error')

    @unittest.skipIf(not hasattr(faulthandler, '_sigill'),
                     "need faulthandler._sigill()")
    def test_sigill(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigill()
""".strip(),
            3,
            'Illegal instruction')

    def test_fatal_error(self):
        self.check_fatal_error("""
import faulthandler
faulthandler._fatal_error(b'xyz')
""".strip(),
            2,
            'xyz')

    @unittest.skipIf(not hasattr(faulthandler, '_stack_overflow'),
                     'need faulthandler._stack_overflow()')
    def test_stack_overflow(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._stack_overflow()
""".strip(),
            3,
            '(?:Segmentation fault|Bus error)')

    def test_gil_released(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._read_null(True)
""".strip(),
            3,
            '(?:Segmentation fault|Bus error)')

    def test_enable_file(self):
        with temporary_filename() as filename:
            self.check_fatal_error("""
import faulthandler
output = open({filename}, 'wb')
faulthandler.enable(output)
faulthandler._read_null(True)<commit_msg>Issue #11393: Disable test_stack_overflow of test_faulthandler<commit_after>from contextlib import contextmanager
import faulthandler
import re
import signal
import subprocess
import sys
from test import support, script_helper
import tempfile
import unittest

try:
    from resource import setrlimit, RLIMIT_CORE, error as resource_error
except ImportError:
    prepare_subprocess = None
else:
    def prepare_subprocess():
        # don't create core file
        try:
            setrlimit(RLIMIT_CORE, (0, 0))
        except (ValueError, resource_error):
            pass

def expected_traceback(lineno1, lineno2, header, count=1):
    regex = header
    regex += r'  File "\<string\>", line %s in func\n' % lineno1
    regex += r'  File "\<string\>", line %s in \<module\>' % lineno2
    if count != 1:
        regex = (regex + '\n') * (count - 1) + regex
    return '^' + regex + '$'

@contextmanager
def temporary_filename():
    filename = tempfile.mktemp()
    try:
        yield filename
    finally:
        support.unlink(filename)

class FaultHandlerTests(unittest.TestCase):
    def get_output(self, code, expect_success, filename=None):
        """
        Run the specified code in Python (in a new child process) and read the
        output from the standard error or from a file (if filename is set).
        Return the output lines as a list.

        Strip the reference count from the standard error for Python debug
        build, and replace "Current thread 0x00007f8d8fbd9700" by "Current
        thread XXX".
        """
        options = {}
        if prepare_subprocess:
            options['preexec_fn'] = prepare_subprocess
        process = script_helper.spawn_python('-c', code, **options)
        stdout, stderr = process.communicate()
        exitcode = process.wait()
        if expect_success:
            self.assertEqual(exitcode, 0)
        else:
            self.assertNotEqual(exitcode, 0)
        if filename:
            with open(filename, "rb") as fp:
                output = fp.read()
        else:
            output = support.strip_python_stderr(stdout)
        output = output.decode('ascii', 'backslashreplace')
        output = re.sub('Current thread 0x[0-9a-f]+',
                        'Current thread XXX',
                        output)
        return output.splitlines()

    def check_fatal_error(self, code, line_number, name_regex,
                               filename=None, all_threads=False):
        """
        Check that the fault handler for fatal errors is enabled and check the
        traceback from the child process output.

        Raise an error if the output doesn't match the expected format.
        """
        if all_threads:
            header = 'Current thread XXX'
        else:
            header = 'Traceback (most recent call first)'
        regex = """
^Fatal Python error: {name}

{header}:
  File "<string>", line {lineno} in <module>$
""".strip()
        regex = regex.format(
            lineno=line_number,
            name=name_regex,
            header=re.escape(header))
        output = self.get_output(code, False, filename)
        output = '\n'.join(output)
        self.assertRegex(output, regex)

    def test_read_null(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._read_null()
""".strip(),
            3,
            '(?:Segmentation fault|Bus error)')

    def test_sigsegv(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigsegv()
""".strip(),
            3,
            'Segmentation fault')

    @unittest.skipIf(sys.platform == 'win32',
                     "SIGFPE cannot be caught on Windows")
    def test_sigfpe(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigfpe()
""".strip(),
            3,
            'Floating point exception')

    @unittest.skipIf(not hasattr(faulthandler, '_sigbus'),
                     "need faulthandler._sigbus()")
    def test_sigbus(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigbus()
""".strip(),
            3,
            'Bus error')

    @unittest.skipIf(not hasattr(faulthandler, '_sigill'),
                     "need faulthandler._sigill()")
    def test_sigill(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._sigill()
""".strip(),
            3,
            'Illegal instruction')

    def test_fatal_error(self):
        self.check_fatal_error("""
import faulthandler
faulthandler._fatal_error(b'xyz')
""".strip(),
            2,
            'xyz')

    @unittest.skipIf(True, 'test disabled, see #11393')
    @unittest.skipIf(not hasattr(faulthandler, '_stack_overflow'),
                     'need faulthandler._stack_overflow()')
    def test_stack_overflow(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._stack_overflow()
""".strip(),
            3,
            '(?:Segmentation fault|Bus error)')

    def test_gil_released(self):
        self.check_fatal_error("""
import faulthandler
faulthandler.enable()
faulthandler._read_null(True)
""".strip(),
            3,
            '(?:Segmentation fault|Bus error)')

    def test_enable_file(self):
        with temporary_filename() as filename:
            self.check_fatal_error("""
import faulthandler
output = open({filename}, 'wb')
faulthandler.enable(output)
faulthandler._read_null(True)
{'fraction_non_alphanumeric': 0.10962542357358733, 'fraction_numerical': 0.007509845223921605, 'mean_word_length': 2.4932821497120923, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '3840128', 'n_tokens_mistral': 3357, 'n_tokens_neox': 3232, 'n_words': 827}