Repository: django/django
File: tests/test_runner/test_parallel.py (Python)

Code before:
        result = TestResult()
        test = TestCase()
        err = _test_error_exc_info()
        event = ("addFailure", 0, err)
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.failures), 1)
        actual_test, tb_and_details_str = result.failures[0]
        self.assertIsInstance(actual_test, TestCase)
        self.assertEqual(actual_test.id(), "unittest.case.TestCase.runTest")
        self.assertIn("Traceback (most recent call last):", tb_and_details_str)
        self.assertIn("ValueError: woops", tb_and_details_str)

    def test_handle_add_success(self):
        dummy_subsuites = []
        pts = ParallelTestSuite(dummy_subsuites, processes=2)
        result = TestResult()
        test = TestCase()
        event = ("addSuccess", 0)
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.failures), 0)


class ParallelBufferTest(DjangoTestCase):
    def test_parallel_buffer(self):
        """
        Test that ParallelTestSuite.run() properly handles buffered TestResult
        to prevent AttributeError when setUpTestData raises exceptions.
        Before the fix, running tests with
        --parallel --buffer would crash with AttributeError when setUpTestData
        raised an exception because the main process TestResult had buffer=True
        but _setupStdout() was never called.
        """

        class AbortingTest(DjangoTestCase):
            @classmethod
            def setUpTestData(cls):
                raise RuntimeError("Boo!")

            def test_pass(self):
                pass

        test = AbortingTest("test_pass")
        remote_result = RemoteTestResult()
        suite = TestSuite([test])
        suite.run(remote_result)

        pts = ParallelTestSuite([suite], processes=2, buffer=True)
        pts.serialized_aliases = set()

Review comment (suggestion):
I think the test would fit better inside the existing `ParallelTestSuiteTest` class on line 221. If you move to `setUpClass`, per the below comment, you won't need Django's `TestCase` to unblock database queries.

Code after:
        test = TestCase()
        err = _test_error_exc_info()
        event = ("addFailure", 0, err)
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.failures), 1)
        actual_test, tb_and_details_str = result.failures[0]
        self.assertIsInstance(actual_test, TestCase)
        self.assertEqual(actual_test.id(), "unittest.case.TestCase.runTest")
        self.assertIn("Traceback (most recent call last):", tb_and_details_str)
        self.assertIn("ValueError: woops", tb_and_details_str)

    def test_handle_add_success(self):
        dummy_subsuites = []
        pts = ParallelTestSuite(dummy_subsuites, processes=2)
        result = TestResult()
        test = TestCase()
        event = ("addSuccess", 0)
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.failures), 0)

    def test_buffer_mode_reports_setupclass_failure(self):
        test = SampleErrorTest("dummy_test")
        remote_result = RemoteTestResult()
        suite = TestSuite([test])
        suite.run(remote_result)

        pts = ParallelTestSuite([suite], processes=2, buffer=True)
        pts.serialized_aliases = set()
        test_result = TestResult()
        test_result.buffer = True

        with unittest.mock.patch("multiprocessing.Pool") as mock_pool:

            def fake_next(*args, **kwargs):
                test_result.shouldStop = True
                return (0, remote_result.events)

            mock_pool.return_value.imap_unordered.return_value = unittest.mock.Mock(
                next=fake_next
            )
            pts.run(test_result)

        self.assertIn("ValueError: woops", test_result.errors[0][1])