1. from django.db import DEFAULT_DB_ALIAS, connections
    
  2. from django.test import LiveServerTestCase, TransactionTestCase
    
  3. from django.test.testcases import LiveServerThread
    
  4. 
    
  5. 
    
  6. # Use TransactionTestCase instead of TestCase to run outside of a transaction,
    
  7. # otherwise closing the connection would implicitly rollback and not set the
    
  8. # connection to None.
    
  9. class LiveServerThreadTest(TransactionTestCase):
    
  10.     available_apps = []
    
  11. 
    
  12.     def run_live_server_thread(self, connections_override=None):
    
  13.         thread = LiveServerTestCase._create_server_thread(connections_override)
    
  14.         thread.daemon = True
    
  15.         thread.start()
    
  16.         thread.is_ready.wait()
    
  17.         thread.terminate()
    
  18. 
    
  19.     def test_closes_connections(self):
    
  20.         conn = connections[DEFAULT_DB_ALIAS]
    
  21.         # Pass a connection to the thread to check they are being closed.
    
  22.         connections_override = {DEFAULT_DB_ALIAS: conn}
    
  23.         # Open a connection to the database.
    
  24.         conn.connect()
    
  25.         conn.inc_thread_sharing()
    
  26.         try:
    
  27.             self.assertIsNotNone(conn.connection)
    
  28.             self.run_live_server_thread(connections_override)
    
  29.             self.assertIsNone(conn.connection)
    
  30.         finally:
    
  31.             conn.dec_thread_sharing()
    
  32. 
    
  33.     def test_server_class(self):
    
  34.         class FakeServer:
    
  35.             def __init__(*args, **kwargs):
    
  36.                 pass
    
  37. 
    
  38.         class MyServerThread(LiveServerThread):
    
  39.             server_class = FakeServer
    
  40. 
    
  41.         class MyServerTestCase(LiveServerTestCase):
    
  42.             server_thread_class = MyServerThread
    
  43. 
    
  44.         thread = MyServerTestCase._create_server_thread(None)
    
  45.         server = thread._create_server()
    
  46.         self.assertIs(type(server), FakeServer)