Fixed #33277 -- Disallowed database connections in threads in SimpleTestCase.

1 files changed, 32 insertions, 0 deletions

diff --git a/tests/test_utils/tests.py b/tests/test_utils/tests.py
index ce78ffc008..65a782bf87 100644
--- a/tests/test_utils/tests.py
+++ b/tests/test_utils/tests.py
@@ -1,5 +1,6 @@
 import os
 import sys
+import threading
 import unittest
 import warnings
 from io import StringIO
@@ -2093,6 +2094,29 @@ class DisallowedDatabaseQueriesTests(SimpleTestCase):
         with self.assertRaisesMessage(DatabaseOperationForbidden, expected_message):
             next(Car.objects.iterator())
 
+    def test_disallowed_thread_database_connection(self):
+        expected_message = (
+            "Database threaded connections to 'default' are not allowed in "
+            "SimpleTestCase subclasses. Either subclass TestCase or TransactionTestCase"
+            " to ensure proper test isolation or add 'default' to "
+            "test_utils.tests.DisallowedDatabaseQueriesTests.databases to "
+            "silence this failure."
+        )
+
+        exceptions = []
+
+        def thread_func():
+            try:
+                Car.objects.first()
+            except DatabaseOperationForbidden as e:
+                exceptions.append(e)
+
+        t = threading.Thread(target=thread_func)
+        t.start()
+        t.join()
+        self.assertEqual(len(exceptions), 1)
+        self.assertEqual(exceptions[0].args[0], expected_message)
+
 
 class AllowedDatabaseQueriesTests(SimpleTestCase):
     databases = {"default"}
@@ -2103,6 +2127,14 @@ class AllowedDatabaseQueriesTests(SimpleTestCase):
     def test_allowed_database_chunked_cursor_queries(self):
         next(Car.objects.iterator(), None)
 
+    def test_allowed_threaded_database_queries(self):
+        def thread_func():
+            next(Car.objects.iterator(), None)
+
+        t = threading.Thread(target=thread_func)
+        t.start()
+        t.join()
+
 
 class DatabaseAliasTests(SimpleTestCase):
     def setUp(self):