tests for django core servers this event is set right after the first time a request closes its database connections set the connection s connmaxage to none to simulate the connmaxage setting being set to none on the server this prevents django from closing the connection and allows testing that threadedwsgiserver closes connections pass a database connection through to the server to check it is being closed by threadedwsgiserver the server s request thread sets this event after closing its database connections open a connection to the database the server can access the database wait for the server s request thread to close the connection a timeout of 0 1 seconds should be more than enough if the wait times out the assertion after should fail with a threaded liveserver and an inmemory database an error can occur when 2 requests reach the server and try to lock the database at the same time if the requests do not share the same database connection open a connection to the database create a transaction to lock the database release the transaction liveservertestcase s change to allowedhosts should be reverted put it in a list to prevent descriptor lookups in test launched server serves with http 1 1 with self urlopenexampleview as f self assertequalf version 11 def testclosesconnectionwithoutcontentlengthself conn httpconnection liveserverviews serverthread host liveserverviews serverthread port timeout1 try conn request get streamingexampleview headersconnection keepalive response conn getresponse self asserttrueresponse willclose self assertequalresponse read biamastream self assertequalresponse status 200 self assertequalresponse getheaderconnection close conn request get streamingexampleview headersconnection close response conn getresponse self asserttrueresponse willclose self assertequalresponse read biamastream self assertequalresponse status 200 self assertequalresponse getheaderconnection close finally conn close def testkeepaliveonconnectionwithcontentlengthself conn httpconnection liveserverviews serverthread host liveserverviews serverthread port try conn requestget exampleview headersconnection keepalive response conn getresponse self assertfalseresponse willclose self assertequalresponse read bexample view self assertequalresponse status 200 self assertisnoneresponse getheaderconnection conn requestget exampleview headersconnection close response conn getresponse self assertfalseresponse willclose self assertequalresponse read bexample view self assertequalresponse status 200 self assertisnoneresponse getheaderconnection finally conn close def testkeepaliveconnectionclearspreviousrequestdataself conn httpconnection liveserverviews serverthread host liveserverviews serverthread port try conn request post methodview b headersconnection keepalive response conn getresponse self assertfalseresponse willclose self assertequalresponse status 200 self assertequalresponse read bpost conn request post methodview b headersconnection close response conn getresponse self assertfalseresponse willclose self assertequalresponse status 200 self assertequalresponse read bpost finally conn close def test404self with self assertraiseshttperror as err self urlopen err exception close self assertequalerr exception code 404 expected 404 response def testviewself with self urlopenexampleview as f self assertequalf read bexample view def teststaticfilesself with self urlopenstaticexamplestaticfile txt as f self assertequalf read rstripbrn bexample static file def testnocollectstaticemulationself with self assertraiseshttperror as err self urlopenstaticanotherappanotherappstaticfile txt err exception close self assertequalerr exception code 404 expected 404 response def testmediafilesself with self urlopenmediaexamplemediafile txt as f self assertequalf read rstripbrn bexample media file def testenvironself with self urlopenenvironview s urlencodeq as f self assertinbquerystring qd182d0b5d181d182 f read overridesettingsrooturlconfservers urls class singlethreadliveserverviewssinglethreadliveservertestcase availableapps servers def testclosesconnectionwithcontentlengthself conn httpconnection singlethreadliveserverviews serverthread host singlethreadliveserverviews serverthread port timeout1 try conn requestget exampleview headersconnection keepalive response conn getresponse self asserttrueresponse willclose self assertequalresponse read bexample view self assertequalresponse status 200 self assertequalresponse getheaderconnection close finally conn close class liveserverdatabaseliveserverbase def testfixturesloadedself with self urlopenmodelview as f self assertcountequalf read splitlines bjane brobert def testdatabasewritesself with self urlopencreatemodelinstance pass self assertquerysetequal person objects orderbypk jane robert emily lambda b b name class liveserverportliveserverbase def testportbindself testcase typetestcase liveserverbase try testcase startserverthread except oserror as e if e errno errno eaddrinuse we re out of ports liveservertestcase correctly fails with an oserror return unexpected error raise try self assertnotequal self liveserverurl testcase liveserverurl facquired duplicate server addresses for server threads fself liveserverurl finally testcase doclasscleanups def testspecifiedportbindself find an open port and tell testcase to use it if liveservertestcase isn t threaded these tests will hang def testviewcallssubviewself url subviewcallingview s urlencodeurl self liveserverurl with self urlopenurl as f self assertequalf read bsubview calling view subview def testcheckmodelinstancefromsubviewself url checkmodelinstancefromsubview s urlencode url self liveserverurl with self urlopenurl as f self assertinbemily f read this event is set right after the first time a request closes its database connections set the connection s conn_max_age to none to simulate the conn_max_age setting being set to none on the server this prevents django from closing the connection and allows testing that threadedwsgiserver closes connections pass a database connection through to the server to check it is being closed by threadedwsgiserver the server s request thread sets this event after closing its database connections open a connection to the database the server can access the database wait for the server s request thread to close the connection a timeout of 0 1 seconds should be more than enough if the wait times out the assertion after should fail with a threaded liveserver and an in memory database an error can occur when 2 requests reach the server and try to lock the database at the same time if the requests do not share the same database connection open a connection to the database create a transaction to lock the database release the transaction liveservertestcase s change to allowed_hosts should be reverted put it in a list to prevent descriptor lookups in test launched server serves with http 1 1 an http 1 1 server is supposed to support keep alive since our development server is rather simple we support it only in cases where we can detect a content length from the response this should be doable for all simple views and streaming responses where an iterable with length of one is passed the latter follows as result of set_content_length from https github com python cpython blob main lib wsgiref handlers py if we cannot detect a content length we explicitly set the connection header to close to notify the client that we do not actually support it see test_closes_connection_without_content_length for details this is a follow up test which ensure that we do not close the connection if not needed hence allowing us to take advantage of keep alive liveservertestcase reports a 404 status code when http client tries to access a static file that isn t explicitly put under static_root contrast to liveserverviews test_keep_alive_on_connection_with_content_length persistent connections require threading server fixtures are properly loaded and visible to the live server thread data written to the database by a view can be read each liveservertestcase binds to a unique port or fails to start a server thread when run concurrently 26011 we re out of ports liveservertestcase correctly fails with an oserror unexpected error liveservertestcase port customizes the server s port find an open port and tell testcase to use it if liveservertestcase isn t threaded these tests will hang
import errno
import os
import socket
import threading
import unittest
from http.client import HTTPConnection
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import urlopen

from django.conf import settings
from django.core.servers.basehttp import ThreadedWSGIServer, WSGIServer
from django.db import DEFAULT_DB_ALIAS, connection, connections
from django.test import LiveServerTestCase, override_settings
from django.test.testcases import LiveServerThread, QuietWSGIRequestHandler

from .models import Person

TEST_ROOT = os.path.dirname(__file__)
TEST_SETTINGS = {
    "MEDIA_URL": "media/",
    "MEDIA_ROOT": os.path.join(TEST_ROOT, "media"),
    "STATIC_URL": "static/",
    "STATIC_ROOT": os.path.join(TEST_ROOT, "static"),
}


@override_settings(ROOT_URLCONF="servers.urls", **TEST_SETTINGS)
class LiveServerBase(LiveServerTestCase):
    available_apps = [
        "servers",
        "django.contrib.auth",
        "django.contrib.contenttypes",
        "django.contrib.sessions",
    ]
    fixtures = ["testdata.json"]

    def urlopen(self, url):
        return urlopen(self.live_server_url + url)


class CloseConnectionTestServer(ThreadedWSGIServer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        
        self._connections_closed = threading.Event()

    def _close_connections(self):
        super()._close_connections()
        self._connections_closed.set()


class CloseConnectionTestLiveServerThread(LiveServerThread):
    server_class = CloseConnectionTestServer

    def _create_server(self, connections_override=None):
        return super()._create_server(connections_override=self.connections_override)


class LiveServerTestCloseConnectionTest(LiveServerBase):
    server_thread_class = CloseConnectionTestLiveServerThread

    @classmethod
    def _make_connections_override(cls):
        conn = connections[DEFAULT_DB_ALIAS]
        cls.conn = conn
        cls.old_conn_max_age = conn.settings_dict["CONN_MAX_AGE"]
        
        
        
        
        conn.settings_dict["CONN_MAX_AGE"] = None
        
        
        return {DEFAULT_DB_ALIAS: conn}

    @classmethod
    def tearDownConnectionTest(cls):
        cls.conn.settings_dict["CONN_MAX_AGE"] = cls.old_conn_max_age

    @classmethod
    def tearDownClass(cls):
        cls.tearDownConnectionTest()
        super().tearDownClass()

    def test_closes_connections(self):
        
        
        closed_event = self.server_thread.httpd._connections_closed
        conn = self.conn
        
        conn.connect()
        self.assertIsNotNone(conn.connection)
        with self.urlopen("/model_view/") as f:
            
            self.assertCountEqual(f.read().splitlines(), [b"jane", b"robert"])
        
        
        
        closed_event.wait(timeout=0.1)
        self.assertIsNone(conn.connection)


@unittest.skipUnless(connection.vendor == "sqlite", "SQLite specific test.")
class LiveServerInMemoryDatabaseLockTest(LiveServerBase):
    def test_in_memory_database_lock(self):
        
        conn = self.server_thread.connections_override[DEFAULT_DB_ALIAS]
        source_connection = conn.connection
        
        conn.connect()
        
        cursor = conn.cursor()
        cursor.execute("BEGIN IMMEDIATE TRANSACTION")
        try:
            with self.urlopen("/create_model_instance/") as f:
                self.assertEqual(f.status, 200)
        except HTTPError:
            self.fail("Unexpected error due to a database lock.")
        finally:
            
            cursor.execute("ROLLBACK")
            source_connection.close()


class FailingLiveServerThread(LiveServerThread):
    def _create_server(self, connections_override=None):
        raise RuntimeError("Error creating server.")


class LiveServerTestCaseSetupTest(LiveServerBase):
    server_thread_class = FailingLiveServerThread

    @classmethod
    def check_allowed_hosts(cls, expected):
        if settings.ALLOWED_HOSTS != expected:
            raise RuntimeError(f"{settings.ALLOWED_HOSTS} != {expected}")

    @classmethod
    def setUpClass(cls):
        cls.check_allowed_hosts(["testserver"])
        try:
            super().setUpClass()
        except RuntimeError:
            
            cls.doClassCleanups()
            cls.check_allowed_hosts(["testserver"])
        else:
            raise RuntimeError("Server did not fail.")
        cls.set_up_called = True

    def test_set_up_class(self):
        self.assertIs(self.set_up_called, True)


class LiveServerAddress(LiveServerBase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        
        cls.live_server_url_test = [cls.live_server_url]

    def test_live_server_url_is_class_property(self):
        self.assertIsInstance(self.live_server_url_test[0], str)
        self.assertEqual(self.live_server_url_test[0], self.live_server_url)


class LiveServerSingleThread(LiveServerThread):
    def _create_server(self, connections_override=None):
        return WSGIServer(
            (self.host, self.port), QuietWSGIRequestHandler, allow_reuse_address=False
        )


class SingleThreadLiveServerTestCase(LiveServerTestCase):
    server_thread_class = LiveServerSingleThread


class LiveServerViews(LiveServerBase):
    def test_protocol(self):
        
        with self.urlopen("/example_view/") as f:
            self.assertEqual(f.version, 11)

    def test_closes_connection_without_content_length(self):
        
        conn = HTTPConnection(
            LiveServerViews.server_thread.host,
            LiveServerViews.server_thread.port,
            timeout=1,
        )
        try:
            conn.request(
                "GET", "/streaming_example_view/", headers={"Connection": "keep-alive"}
            )
            response = conn.getresponse()
            self.assertTrue(response.will_close)
            self.assertEqual(response.read(), b"Iamastream")
            self.assertEqual(response.status, 200)
            self.assertEqual(response.getheader("Connection"), "close")

            conn.request(
                "GET", "/streaming_example_view/", headers={"Connection": "close"}
            )
            response = conn.getresponse()
            self.assertTrue(response.will_close)
            self.assertEqual(response.read(), b"Iamastream")
            self.assertEqual(response.status, 200)
            self.assertEqual(response.getheader("Connection"), "close")
        finally:
            conn.close()

    def test_keep_alive_on_connection_with_content_length(self):
        
        conn = HTTPConnection(
            LiveServerViews.server_thread.host, LiveServerViews.server_thread.port
        )
        try:
            conn.request("GET", "/example_view/", headers={"Connection": "keep-alive"})
            response = conn.getresponse()
            self.assertFalse(response.will_close)
            self.assertEqual(response.read(), b"example view")
            self.assertEqual(response.status, 200)
            self.assertIsNone(response.getheader("Connection"))

            conn.request("GET", "/example_view/", headers={"Connection": "close"})
            response = conn.getresponse()
            self.assertFalse(response.will_close)
            self.assertEqual(response.read(), b"example view")
            self.assertEqual(response.status, 200)
            self.assertIsNone(response.getheader("Connection"))
        finally:
            conn.close()

    def test_keep_alive_connection_clears_previous_request_data(self):
        conn = HTTPConnection(
            LiveServerViews.server_thread.host, LiveServerViews.server_thread.port
        )
        try:
            conn.request(
                "POST", "/method_view/", b"{}", headers={"Connection": "keep-alive"}
            )
            response = conn.getresponse()
            self.assertFalse(response.will_close)
            self.assertEqual(response.status, 200)
            self.assertEqual(response.read(), b"POST")

            conn.request(
                "POST", "/method_view/", b"{}", headers={"Connection": "close"}
            )
            response = conn.getresponse()
            self.assertFalse(response.will_close)
            self.assertEqual(response.status, 200)
            self.assertEqual(response.read(), b"POST")
        finally:
            conn.close()

    def test_404(self):
        with self.assertRaises(HTTPError) as err:
            self.urlopen("/")
        err.exception.close()
        self.assertEqual(err.exception.code, 404, "Expected 404 response")

    def test_view(self):
        with self.urlopen("/example_view/") as f:
            self.assertEqual(f.read(), b"example view")

    def test_static_files(self):
        with self.urlopen("/static/example_static_file.txt") as f:
            self.assertEqual(f.read().rstrip(b"\r\n"), b"example static file")

    def test_no_collectstatic_emulation(self):
        
        with self.assertRaises(HTTPError) as err:
            self.urlopen("/static/another_app/another_app_static_file.txt")
        err.exception.close()
        self.assertEqual(err.exception.code, 404, "Expected 404 response")

    def test_media_files(self):
        with self.urlopen("/media/example_media_file.txt") as f:
            self.assertEqual(f.read().rstrip(b"\r\n"), b"example media file")

    def test_environ(self):
        with self.urlopen("/environ_view/?%s" % urlencode({"q": "тест"})) as f:
            self.assertIn(b"QUERY_STRING: 'q=%D1%82%D0%B5%D1%81%D1%82'", f.read())


@override_settings(ROOT_URLCONF="servers.urls")
class SingleThreadLiveServerViews(SingleThreadLiveServerTestCase):
    available_apps = ["servers"]

    def test_closes_connection_with_content_length(self):
        
        conn = HTTPConnection(
            SingleThreadLiveServerViews.server_thread.host,
            SingleThreadLiveServerViews.server_thread.port,
            timeout=1,
        )
        try:
            conn.request("GET", "/example_view/", headers={"Connection": "keep-alive"})
            response = conn.getresponse()
            self.assertTrue(response.will_close)
            self.assertEqual(response.read(), b"example view")
            self.assertEqual(response.status, 200)
            self.assertEqual(response.getheader("Connection"), "close")
        finally:
            conn.close()


class LiveServerDatabase(LiveServerBase):
    def test_fixtures_loaded(self):
        
        with self.urlopen("/model_view/") as f:
            self.assertCountEqual(f.read().splitlines(), [b"jane", b"robert"])

    def test_database_writes(self):
        
        with self.urlopen("/create_model_instance/"):
            pass
        self.assertQuerySetEqual(
            Person.objects.order_by("pk"),
            ["jane", "robert", "emily"],
            lambda b: b.name,
        )


class LiveServerPort(LiveServerBase):
    def test_port_bind(self):
        
        TestCase = type("TestCase", (LiveServerBase,), {})
        try:
            TestCase._start_server_thread()
        except OSError as e:
            if e.errno == errno.EADDRINUSE:
                
                
                return
            
            raise
        try:
            self.assertNotEqual(
                self.live_server_url,
                TestCase.live_server_url,
                f"Acquired duplicate server addresses for server threads: "
                f"{self.live_server_url}",
            )
        finally:
            TestCase.doClassCleanups()

    def test_specified_port_bind(self):
        
        TestCase = type("TestCase", (LiveServerBase,), {})
        
        s = socket.socket()
        s.bind(("", 0))
        TestCase.port = s.getsockname()[1]
        s.close()
        TestCase._start_server_thread()
        try:
            self.assertEqual(
                TestCase.port,
                TestCase.server_thread.port,
                f"Did not use specified port for LiveServerTestCase thread: "
                f"{TestCase.port}",
            )
        finally:
            TestCase.doClassCleanups()


class LiveServerThreadedTests(LiveServerBase):
    

    def test_view_calls_subview(self):
        url = "/subview_calling_view/?%s" % urlencode({"url": self.live_server_url})
        with self.urlopen(url) as f:
            self.assertEqual(f.read(), b"subview calling view: subview")

    def test_check_model_instance_from_subview(self):
        url = "/check_model_instance_from_subview/?%s" % urlencode(
            {
                "url": self.live_server_url,
            }
        )
        with self.urlopen(url) as f:
            self.assertIn(b"emily", f.read())