Ensured that thread-shareability gets validated when closing a PostgreSQL or SQLite connection. Refs #17258.

git-svn-id: http://code.djangoproject.com/svn/django/trunk@17206 bcc190cf-cafb-0310-a4f2-bffc1f526a37

1 files changed, 45 insertions, 0 deletions

diff --git a/tests/regressiontests/backends/tests.py b/tests/regressiontests/backends/tests.py
index 82c21c8c7b..bda604e00b 100644
--- a/tests/regressiontests/backends/tests.py
+++ b/tests/regressiontests/backends/tests.py
@@ -487,6 +487,9 @@ class ThreadTests(TestCase):
         def runner():
             from django.db import connections
             for conn in connections.all():
+                # Allow thread sharing so the connection can be closed by the
+                # main thread.
+                conn.allow_thread_sharing = True
                 connections_set.add(conn)
         for x in xrange(2):
             t = threading.Thread(target=runner)
@@ -537,4 +540,46 @@ class ThreadTests(TestCase):
         exceptions = []
         do_thread()
         # All good
+        self.assertEqual(len(exceptions), 0)
+
+    def test_closing_non_shared_connections(self):
+        """
+        Ensure that a connection that is not explicitly shareable cannot be
+        closed by another thread.
+        Refs #17258.
+        """
+        # First, without explicitly enabling the connection for sharing.
+        exceptions = set()
+        def runner1():
+            def runner2(other_thread_connection):
+                try:
+                    other_thread_connection.close()
+                except DatabaseError, e:
+                    exceptions.add(e)
+            t2 = threading.Thread(target=runner2, args=[connections['default']])
+            t2.start()
+            t2.join()
+        t1 = threading.Thread(target=runner1)
+        t1.start()
+        t1.join()
+        # The exception was raised
+        self.assertEqual(len(exceptions), 1)
+
+        # Then, with explicitly enabling the connection for sharing.
+        exceptions = set()
+        def runner1():
+            def runner2(other_thread_connection):
+                try:
+                    other_thread_connection.close()
+                except DatabaseError, e:
+                    exceptions.add(e)
+            # Enable thread sharing
+            connections['default'].allow_thread_sharing = True
+            t2 = threading.Thread(target=runner2, args=[connections['default']])
+            t2.start()
+            t2.join()
+        t1 = threading.Thread(target=runner1)
+        t1.start()
+        t1.join()
+        # No exception was raised
         self.assertEqual(len(exceptions), 0)
\ No newline at end of file