<commit_before>from __future__ import unicode_literals
import copy
import os
import platform
import shutil
import signal
import sys
import tempfile
import unittest
import weakref

from green.config import default_args
from green.loader import loadTargets
from green.output import GreenStream
from green.runner import run
from green.suite import GreenTestSuite

try:
    from io import StringIO
except:
    from StringIO import StringIO


class FakeCase(unittest.TestCase):
    def runTest(self):
        pass


class TestRun(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.startdir = os.getcwd()

    @classmethod
    def tearDownClass(cls):
        if os.getcwd() != cls.startdir:
            os.chdir(cls.startdir)
        cls.startdir = None

    def setUp(self):
        self.args = copy.deepcopy(default_args)
        self.stream = StringIO()
        self.tmpdir = tempfile.mkdtemp()

    def tearDown(self):
        del(self.tmpdir)
        del(self.stream)

    def test_catchSIGINT(self):
        """
        run() can catch SIGINT with just one process.
        """
        if platform.system() == 'Windows':
            self.skipTest('This test is for posix-specific behavior.')
        # Mock the list of TestResult instances that should be stopped,
        # otherwise the actual TestResult that is running this test will be
        # told to stop when we send SIGINT
        saved__results = unittest.signals._results
        unittest.signals._results = weakref.WeakKeyDictionary()
        class KBICase(unittest.TestCase):
            def runTest(self):
                os.kill(os.getpid(), signal.SIGINT)
        kc = KBICase()
        run(kc, self.stream, self.args)
        unittest.signals._results = saved__results

    def test_stdout(self):
        """
        run() can use sys.stdout as the stream.
        """
        saved_stdout = sys.stdout
        sys.stdout = self.stream
        run(GreenTestSuite(), sys.stdout, args=self.args)
        sys.stdout = saved_stdout
        self.assertIn('No Tests Found', self.stream.getvalue())

    def test_GreenStream(self):
        """
        run() can use a GreenStream for output.
        """
        gs = GreenStream(self.stream)
        run(GreenTestSuite(), gs, args=self.args)
        self.assertIn('No Tests Found', self.stream.getvalue())

    def test_HTML(self):
        """
        html=True causes html output
        """
        self.args.html = True
        run(FakeCase(), self.stream, self.args)
        self.assertIn('<', self.stream.getvalue())

    def test_verbose3(self):
        """
        verbose=3 causes version output, and an empty test case passes.
        """
        self.args.verbose = 3
        run(FakeCase(), self.stream, self.args)
        self.assertIn('Green', self.stream.getvalue())
        self.assertIn('OK', self.stream.getvalue())

    def test_warnings(self):
        """
        setting warnings='always' doesn't crash
        """
        self.args.warnings = 'always'
        run(FakeCase(), self.stream, self.args)
        self.assertIn('OK', self.stream.getvalue())

    def test_noTestsFound(self):
        """
        When we don't find any tests, we say so.
        """
        run(GreenTestSuite(), self.stream, self.args)
        self.assertIn('No Tests Found', self.stream.getvalue())

    def test_failedSaysSo(self):
        """
        A failing test case causes the whole run to report 'FAILED'
        """
        class FailCase(unittest.TestCase):
            def runTest(self):
                self.assertTrue(False)
        run(FailCase(), self.stream, self.args)
        self.assertIn('FAILED', self.stream.getvalue())

    def test_failfast(self):
        """
        failfast causes the testing to stop after the first failure.
        """
        sub_tmpdir = tempfile.mkdtemp(dir=self.tmpdir)
        fh = open(os.path.join(sub_tmpdir, 'test_failfast.py'), 'w')
        fh.write("""
import unittest
class SIGINTCase(unittest.TestCase):
    def test00(self):
        raise Exception
    def test01(self):
        pass
""".format(os.getpid()))
        fh.close()
        os.chdir(sub_tmpdir)
        tests = loadTargets('test_failfast')
        self.args.failfast = True
        result = run(tests, self.stream, self.args)
        os.chdir(self.startdir)
        self.assertEqual(result.testsRun, 1)


class TestSubprocesses(unittest.TestCase):

    # Setup
    @classmethod
    def setUpClass(cls):<commit_msg>Added a test that ensures that the SystemExit can be caught as part of investigating issue #60.  It turned out that the issue was actually a design decision by the testtools devs and happens when you subclass testtools.TestCase, so this test already passed and I didn't change anything.  But I figured it wouldn't help to have one more test that tests something, so I'll leave this one in.<commit_after>from __future__ import unicode_literals
import copy
import os
import platform
import shutil
import signal
import sys
import tempfile
import unittest
import weakref

from green.config import default_args
from green.loader import loadTargets
from green.output import GreenStream
from green.runner import run
from green.suite import GreenTestSuite

try:
    from io import StringIO
except:
    from StringIO import StringIO


class FakeCase(unittest.TestCase):
    def runTest(self):
        pass


class TestRun(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.startdir = os.getcwd()

    @classmethod
    def tearDownClass(cls):
        if os.getcwd() != cls.startdir:
            os.chdir(cls.startdir)
        cls.startdir = None

    def setUp(self):
        self.args = copy.deepcopy(default_args)
        self.stream = StringIO()
        self.tmpdir = tempfile.mkdtemp()

    def tearDown(self):
        del(self.tmpdir)
        del(self.stream)

    def test_catchSIGINT(self):
        """
        run() can catch SIGINT with just one process.
        """
        if platform.system() == 'Windows':
            self.skipTest('This test is for posix-specific behavior.')
        # Mock the list of TestResult instances that should be stopped,
        # otherwise the actual TestResult that is running this test will be
        # told to stop when we send SIGINT
        saved__results = unittest.signals._results
        unittest.signals._results = weakref.WeakKeyDictionary()
        class KBICase(unittest.TestCase):
            def runTest(self):
                os.kill(os.getpid(), signal.SIGINT)
        kc = KBICase()
        run(kc, self.stream, self.args)
        unittest.signals._results = saved__results

    def test_stdout(self):
        """
        run() can use sys.stdout as the stream.
        """
        saved_stdout = sys.stdout
        sys.stdout = self.stream
        run(GreenTestSuite(), sys.stdout, args=self.args)
        sys.stdout = saved_stdout
        self.assertIn('No Tests Found', self.stream.getvalue())

    def test_GreenStream(self):
        """
        run() can use a GreenStream for output.
        """
        gs = GreenStream(self.stream)
        run(GreenTestSuite(), gs, args=self.args)
        self.assertIn('No Tests Found', self.stream.getvalue())

    def test_HTML(self):
        """
        html=True causes html output
        """
        self.args.html = True
        run(FakeCase(), self.stream, self.args)
        self.assertIn('<', self.stream.getvalue())

    def test_verbose3(self):
        """
        verbose=3 causes version output, and an empty test case passes.
        """
        self.args.verbose = 3
        run(FakeCase(), self.stream, self.args)
        self.assertIn('Green', self.stream.getvalue())
        self.assertIn('OK', self.stream.getvalue())

    def test_warnings(self):
        """
        setting warnings='always' doesn't crash
        """
        self.args.warnings = 'always'
        run(FakeCase(), self.stream, self.args)
        self.assertIn('OK', self.stream.getvalue())

    def test_noTestsFound(self):
        """
        When we don't find any tests, we say so.
        """
        run(GreenTestSuite(), self.stream, self.args)
        self.assertIn('No Tests Found', self.stream.getvalue())

    def test_failedSaysSo(self):
        """
        A failing test case causes the whole run to report 'FAILED'
        """
        class FailCase(unittest.TestCase):
            def runTest(self):
                self.assertTrue(False)
        run(FailCase(), self.stream, self.args)
        self.assertIn('FAILED', self.stream.getvalue())

    def test_failfast(self):
        """
        failfast causes the testing to stop after the first failure.
        """
        sub_tmpdir = tempfile.mkdtemp(dir=self.tmpdir)
        fh = open(os.path.join(sub_tmpdir, 'test_failfast.py'), 'w')
        fh.write("""
import unittest
class SIGINTCase(unittest.TestCase):
    def test00(self):
        raise Exception
    def test01(self):
        pass
""".format(os.getpid()))
        fh.close()
        os.chdir(sub_tmpdir)
        tests = loadTargets('test_failfast')
        self.args.failfast = True
        result = run(tests, self.stream, self.args)
        os.chdir(self.startdir)
        self.assertEqual(result.testsRun, 1)


    def test_systemExit(self):
        """
        Raising a SystemExit gets caught and reported.
        """
        sub_tmpdir = tempfile.mkdtemp(dir=self.tmpdir)
        fh = open(os.path.join(sub_tmpdir, 'test_systemexit.py'), 'w')
        fh.write("""
import unittest
class SystemExitCase(unittest.TestCase):
    def test00(self):
        raise SystemExit(1)
    def test01(self):
        pass
""".format(os.getpid()))
        fh.close()
        os.chdir(sub_tmpdir)
        tests = loadTargets('test_systemexit')
        result = run(tests, self.stream, self.args)
        os.chdir(self.startdir)
        self.assertEqual(result.testsRun, 2)



class TestSubprocesses(unittest.TestCase):

    # Setup
    @classmethod
    def setUpClass(cls):
{'fraction_non_alphanumeric': 0.10715360690277917, 'fraction_numerical': 0.0024079462225343636, 'mean_word_length': 2.4136986301369863, 'pattern_counts': {'":': 0, '<': 5, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3275765', 'n_tokens_mistral': 2942, 'n_tokens_neox': 2867, 'n_words': 759}