diff options
Diffstat (limited to 'tests/servers')
| -rw-r--r-- | tests/servers/test_basehttp.py | 66 | ||||
| -rw-r--r-- | tests/servers/tests.py | 173 | ||||
| -rw-r--r-- | tests/servers/urls.py | 18 | ||||
| -rw-r--r-- | tests/servers/views.py | 24 |
4 files changed, 151 insertions, 130 deletions
diff --git a/tests/servers/test_basehttp.py b/tests/servers/test_basehttp.py index b90e8565a6..a6e75a9c48 100644 --- a/tests/servers/test_basehttp.py +++ b/tests/servers/test_basehttp.py @@ -12,46 +12,48 @@ class Stub: self.__dict__.update(kwargs) def sendall(self, data): - self.makefile('wb').write(data) + self.makefile("wb").write(data) class WSGIRequestHandlerTestCase(SimpleTestCase): request_factory = RequestFactory() def test_log_message(self): - request = WSGIRequest(self.request_factory.get('/').environ) + request = WSGIRequest(self.request_factory.get("/").environ) request.makefile = lambda *args, **kwargs: BytesIO() - handler = WSGIRequestHandler(request, '192.168.0.2', None) + handler = WSGIRequestHandler(request, "192.168.0.2", None) level_status_codes = { - 'info': [200, 301, 304], - 'warning': [400, 403, 404], - 'error': [500, 503], + "info": [200, 301, 304], + "warning": [400, 403, 404], + "error": [500, 503], } for level, status_codes in level_status_codes.items(): for status_code in status_codes: # The correct level gets the message. - with self.assertLogs('django.server', level.upper()) as cm: - handler.log_message('GET %s %s', 'A', str(status_code)) - self.assertIn('GET A %d' % status_code, cm.output[0]) + with self.assertLogs("django.server", level.upper()) as cm: + handler.log_message("GET %s %s", "A", str(status_code)) + self.assertIn("GET A %d" % status_code, cm.output[0]) # Incorrect levels don't have any messages. for wrong_level in level_status_codes: if wrong_level != level: - with self.assertLogs('django.server', 'INFO') as cm: - handler.log_message('GET %s %s', 'A', str(status_code)) - self.assertNotEqual(cm.records[0].levelname, wrong_level.upper()) + with self.assertLogs("django.server", "INFO") as cm: + handler.log_message("GET %s %s", "A", str(status_code)) + self.assertNotEqual( + cm.records[0].levelname, wrong_level.upper() + ) def test_https(self): - request = WSGIRequest(self.request_factory.get('/').environ) + request = WSGIRequest(self.request_factory.get("/").environ) request.makefile = lambda *args, **kwargs: BytesIO() - handler = WSGIRequestHandler(request, '192.168.0.2', None) + handler = WSGIRequestHandler(request, "192.168.0.2", None) - with self.assertLogs('django.server', 'ERROR') as cm: - handler.log_message("GET %s %s", '\x16\x03', "4") + with self.assertLogs("django.server", "ERROR") as cm: + handler.log_message("GET %s %s", "\x16\x03", "4") self.assertIn( "You're accessing the development server over HTTPS, " "but it only supports HTTP.", - cm.records[0].getMessage() + cm.records[0].getMessage(), ) def test_strips_underscore_headers(self): @@ -61,14 +63,14 @@ class WSGIRequestHandlerTestCase(SimpleTestCase): ambiguity between dashes and underscores in mapping to WSGI environ, which can have security implications. """ + def test_app(environ, start_response): """A WSGI app that just reflects its HTTP environ.""" - start_response('200 OK', []) + start_response("200 OK", []) http_environ_items = sorted( - '%s:%s' % (k, v) for k, v in environ.items() - if k.startswith('HTTP_') + "%s:%s" % (k, v) for k, v in environ.items() if k.startswith("HTTP_") ) - yield (','.join(http_environ_items)).encode() + yield (",".join(http_environ_items)).encode() rfile = BytesIO() rfile.write(b"GET / HTTP/1.0\r\n") @@ -86,23 +88,23 @@ class WSGIRequestHandlerTestCase(SimpleTestCase): wfile = UnclosableBytesIO() def makefile(mode, *a, **kw): - if mode == 'rb': + if mode == "rb": return rfile - elif mode == 'wb': + elif mode == "wb": return wfile request = Stub(makefile=makefile) server = Stub(base_environ={}, get_app=lambda: test_app) # Prevent logging from appearing in test output. - with self.assertLogs('django.server', 'INFO'): + with self.assertLogs("django.server", "INFO"): # instantiating a handler runs the request as side effect - WSGIRequestHandler(request, '192.168.0.2', server) + WSGIRequestHandler(request, "192.168.0.2", server) wfile.seek(0) body = list(wfile.readlines())[-1] - self.assertEqual(body, b'HTTP_SOME_HEADER:good') + self.assertEqual(body, b"HTTP_SOME_HEADER:good") class WSGIServerTestCase(SimpleTestCase): @@ -110,9 +112,9 @@ class WSGIServerTestCase(SimpleTestCase): def test_broken_pipe_errors(self): """WSGIServer handles broken pipe errors.""" - request = WSGIRequest(self.request_factory.get('/').environ) - client_address = ('192.168.2.0', 8080) - msg = f'- Broken pipe from {client_address}\n' + request = WSGIRequest(self.request_factory.get("/").environ) + client_address = ("192.168.2.0", 8080) + msg = f"- Broken pipe from {client_address}\n" tests = [ BrokenPipeError, ConnectionAbortedError, @@ -121,14 +123,14 @@ class WSGIServerTestCase(SimpleTestCase): for exception in tests: with self.subTest(exception=exception): try: - server = WSGIServer(('localhost', 0), WSGIRequestHandler) + server = WSGIServer(("localhost", 0), WSGIRequestHandler) try: raise exception() except Exception: with captured_stderr() as err: - with self.assertLogs('django.server', 'INFO') as cm: + with self.assertLogs("django.server", "INFO") as cm: server.handle_error(request, client_address) - self.assertEqual(err.getvalue(), '') + self.assertEqual(err.getvalue(), "") self.assertEqual(cm.records[0].getMessage(), msg) finally: server.server_close() diff --git a/tests/servers/tests.py b/tests/servers/tests.py index 2b2eab20aa..0601e1b837 100644 --- a/tests/servers/tests.py +++ b/tests/servers/tests.py @@ -20,23 +20,23 @@ 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'), + "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) +@override_settings(ROOT_URLCONF="servers.urls", **TEST_SETTINGS) class LiveServerBase(LiveServerTestCase): available_apps = [ - 'servers', - 'django.contrib.auth', - 'django.contrib.contenttypes', - 'django.contrib.sessions', + "servers", + "django.contrib.auth", + "django.contrib.contenttypes", + "django.contrib.sessions", ] - fixtures = ['testdata.json'] + fixtures = ["testdata.json"] def urlopen(self, url): return urlopen(self.live_server_url + url) @@ -70,19 +70,19 @@ class LiveServerTestCloseConnectionTest(LiveServerBase): def _make_connections_override(cls): conn = connections[DEFAULT_DB_ALIAS] cls.conn = conn - cls.old_conn_max_age = conn.settings_dict['CONN_MAX_AGE'] + cls.old_conn_max_age = conn.settings_dict["CONN_MAX_AGE"] # 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. - conn.settings_dict['CONN_MAX_AGE'] = None + conn.settings_dict["CONN_MAX_AGE"] = None # Pass a database connection through to the server to check it is being # closed by ThreadedWSGIServer. return {DEFAULT_DB_ALIAS: conn} @classmethod def tearDownConnectionTest(cls): - cls.conn.settings_dict['CONN_MAX_AGE'] = cls.old_conn_max_age + cls.conn.settings_dict["CONN_MAX_AGE"] = cls.old_conn_max_age @classmethod def tearDownClass(cls): @@ -97,9 +97,9 @@ class LiveServerTestCloseConnectionTest(LiveServerBase): # Open a connection to the database. conn.connect() self.assertIsNotNone(conn.connection) - with self.urlopen('/model_view/') as f: + with self.urlopen("/model_view/") as f: # The server can access the database. - self.assertEqual(f.read().splitlines(), [b'jane', b'robert']) + self.assertEqual(f.read().splitlines(), [b"jane", b"robert"]) # 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. @@ -109,7 +109,7 @@ class LiveServerTestCloseConnectionTest(LiveServerBase): class FailingLiveServerThread(LiveServerThread): def _create_server(self): - raise RuntimeError('Error creating server.') + raise RuntimeError("Error creating server.") class LiveServerTestCaseSetupTest(LiveServerBase): @@ -118,19 +118,19 @@ class LiveServerTestCaseSetupTest(LiveServerBase): @classmethod def check_allowed_hosts(cls, expected): if settings.ALLOWED_HOSTS != expected: - raise RuntimeError(f'{settings.ALLOWED_HOSTS} != {expected}') + raise RuntimeError(f"{settings.ALLOWED_HOSTS} != {expected}") @classmethod def setUpClass(cls): - cls.check_allowed_hosts(['testserver']) + cls.check_allowed_hosts(["testserver"]) try: super().setUpClass() except RuntimeError: # LiveServerTestCase's change to ALLOWED_HOSTS should be reverted. cls.doClassCleanups() - cls.check_allowed_hosts(['testserver']) + cls.check_allowed_hosts(["testserver"]) else: - raise RuntimeError('Server did not fail.') + raise RuntimeError("Server did not fail.") cls.set_up_called = True def test_set_up_class(self): @@ -138,7 +138,6 @@ class LiveServerTestCaseSetupTest(LiveServerBase): class LiveServerAddress(LiveServerBase): - @classmethod def setUpClass(cls): super().setUpClass() @@ -152,7 +151,9 @@ class LiveServerAddress(LiveServerBase): class LiveServerSingleThread(LiveServerThread): def _create_server(self): - return WSGIServer((self.host, self.port), QuietWSGIRequestHandler, allow_reuse_address=False) + return WSGIServer( + (self.host, self.port), QuietWSGIRequestHandler, allow_reuse_address=False + ) class SingleThreadLiveServerTestCase(LiveServerTestCase): @@ -162,7 +163,7 @@ class SingleThreadLiveServerTestCase(LiveServerTestCase): class LiveServerViews(LiveServerBase): def test_protocol(self): """Launched server serves with HTTP 1.1.""" - with self.urlopen('/example_view/') as f: + with self.urlopen("/example_view/") as f: self.assertEqual(f.version, 11) def test_closes_connection_without_content_length(self): @@ -178,21 +179,29 @@ class LiveServerViews(LiveServerBase): header to `close` to notify the client that we do not actually support it. """ - conn = HTTPConnection(LiveServerViews.server_thread.host, LiveServerViews.server_thread.port, timeout=1) + conn = HTTPConnection( + LiveServerViews.server_thread.host, + LiveServerViews.server_thread.port, + timeout=1, + ) try: - conn.request('GET', '/streaming_example_view/', headers={'Connection': 'keep-alive'}) + 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.read(), b"Iamastream") self.assertEqual(response.status, 200) - self.assertEqual(response.getheader('Connection'), 'close') + self.assertEqual(response.getheader("Connection"), "close") - conn.request('GET', '/streaming_example_view/', headers={'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.read(), b"Iamastream") self.assertEqual(response.status, 200) - self.assertEqual(response.getheader('Connection'), 'close') + self.assertEqual(response.getheader("Connection"), "close") finally: conn.close() @@ -202,54 +211,62 @@ class LiveServerViews(LiveServerBase): 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. """ - conn = HTTPConnection(LiveServerViews.server_thread.host, LiveServerViews.server_thread.port) + conn = HTTPConnection( + LiveServerViews.server_thread.host, LiveServerViews.server_thread.port + ) try: - conn.request('GET', '/example_view/', headers={"Connection": "keep-alive"}) + 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.read(), b"example view") self.assertEqual(response.status, 200) - self.assertIsNone(response.getheader('Connection')) + self.assertIsNone(response.getheader("Connection")) - conn.request('GET', '/example_view/', headers={"Connection": "close"}) + 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.read(), b"example view") self.assertEqual(response.status, 200) - self.assertIsNone(response.getheader('Connection')) + 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) + conn = HTTPConnection( + LiveServerViews.server_thread.host, LiveServerViews.server_thread.port + ) try: - conn.request('POST', '/method_view/', b'{}', headers={"Connection": "keep-alive"}) + 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') + self.assertEqual(response.read(), b"POST") - conn.request('POST', '/method_view/', b'{}', headers={"Connection": "close"}) + 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') + self.assertEqual(response.read(), b"POST") finally: conn.close() def test_404(self): with self.assertRaises(HTTPError) as err: - self.urlopen('/') + self.urlopen("/") err.exception.close() - self.assertEqual(err.exception.code, 404, 'Expected 404 response') + 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') + 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') + 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): """ @@ -258,22 +275,22 @@ class LiveServerViews(LiveServerBase): STATIC_ROOT. """ with self.assertRaises(HTTPError) as err: - self.urlopen('/static/another_app/another_app_static_file.txt') + self.urlopen("/static/another_app/another_app_static_file.txt") err.exception.close() - self.assertEqual(err.exception.code, 404, 'Expected 404 response') + 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') + 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: + 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') +@override_settings(ROOT_URLCONF="servers.urls") class SingleThreadLiveServerViews(SingleThreadLiveServerTestCase): - available_apps = ['servers'] + available_apps = ["servers"] def test_closes_connection_with_content_length(self): """ @@ -287,40 +304,38 @@ class SingleThreadLiveServerViews(SingleThreadLiveServerTestCase): timeout=1, ) try: - conn.request('GET', '/example_view/', headers={'Connection': 'keep-alive'}) + 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.read(), b"example view") self.assertEqual(response.status, 200) - self.assertEqual(response.getheader('Connection'), 'close') + self.assertEqual(response.getheader("Connection"), "close") finally: conn.close() class LiveServerDatabase(LiveServerBase): - def test_fixtures_loaded(self): """ Fixtures are properly loaded and visible to the live server thread. """ - with self.urlopen('/model_view/') as f: - self.assertEqual(f.read().splitlines(), [b'jane', b'robert']) + with self.urlopen("/model_view/") as f: + self.assertEqual(f.read().splitlines(), [b"jane", b"robert"]) def test_database_writes(self): """ Data written to the database by a view can be read. """ - with self.urlopen('/create_model_instance/'): + with self.urlopen("/create_model_instance/"): pass self.assertQuerysetEqual( - Person.objects.all().order_by('pk'), - ['jane', 'robert', 'emily'], - lambda b: b.name + Person.objects.all().order_by("pk"), + ["jane", "robert", "emily"], + lambda b: b.name, ) class LiveServerPort(LiveServerBase): - def test_port_bind(self): """ Each LiveServerTestCase binds to a unique port or fails to start a @@ -339,24 +354,24 @@ class LiveServerPort(LiveServerBase): self.assertNotEqual( self.live_server_url, TestCase.live_server_url, - f'Acquired duplicate server addresses for server threads: ' - f'{self.live_server_url}', + f"Acquired duplicate server addresses for server threads: " + f"{self.live_server_url}", ) def test_specified_port_bind(self): """LiveServerTestCase.port customizes the server's port.""" - TestCase = type('TestCase', (LiveServerBase,), {}) + TestCase = type("TestCase", (LiveServerBase,), {}) # Find an open port and tell TestCase to use it. s = socket.socket() - s.bind(('', 0)) + s.bind(("", 0)) TestCase.port = s.getsockname()[1] s.close() TestCase._start_server_thread() self.assertEqual( TestCase.port, TestCase.server_thread.port, - f'Did not use specified port for LiveServerTestCase thread: ' - f'{TestCase.port}', + f"Did not use specified port for LiveServerTestCase thread: " + f"{TestCase.port}", ) @@ -364,13 +379,15 @@ class LiveServerThreadedTests(LiveServerBase): """If LiveServerTestCase isn't threaded, these tests will hang.""" def test_view_calls_subview(self): - url = '/subview_calling_view/?%s' % urlencode({'url': self.live_server_url}) + 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') + 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, - }) + 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()) + self.assertIn(b"emily", f.read()) diff --git a/tests/servers/urls.py b/tests/servers/urls.py index d07712776a..54c2dcb834 100644 --- a/tests/servers/urls.py +++ b/tests/servers/urls.py @@ -3,13 +3,13 @@ from django.urls import path from . import views urlpatterns = [ - path('example_view/', views.example_view), - path('streaming_example_view/', views.streaming_example_view), - path('model_view/', views.model_view), - path('create_model_instance/', views.create_model_instance), - path('environ_view/', views.environ_view), - path('subview_calling_view/', views.subview_calling_view), - path('subview/', views.subview), - path('check_model_instance_from_subview/', views.check_model_instance_from_subview), - path('method_view/', views.method_view), + path("example_view/", views.example_view), + path("streaming_example_view/", views.streaming_example_view), + path("model_view/", views.model_view), + path("create_model_instance/", views.create_model_instance), + path("environ_view/", views.environ_view), + path("subview_calling_view/", views.subview_calling_view), + path("subview/", views.subview), + path("check_model_instance_from_subview/", views.check_model_instance_from_subview), + path("method_view/", views.method_view), ] diff --git a/tests/servers/views.py b/tests/servers/views.py index fa30b0ce7b..97a467f0b0 100644 --- a/tests/servers/views.py +++ b/tests/servers/views.py @@ -7,42 +7,44 @@ from .models import Person def example_view(request): - return HttpResponse('example view') + return HttpResponse("example view") def streaming_example_view(request): - return StreamingHttpResponse((b'I', b'am', b'a', b'stream')) + return StreamingHttpResponse((b"I", b"am", b"a", b"stream")) def model_view(request): people = Person.objects.all() - return HttpResponse('\n'.join(person.name for person in people)) + return HttpResponse("\n".join(person.name for person in people)) def create_model_instance(request): - person = Person(name='emily') + person = Person(name="emily") person.save() return HttpResponse() def environ_view(request): - return HttpResponse("\n".join("%s: %r" % (k, v) for k, v in request.environ.items())) + return HttpResponse( + "\n".join("%s: %r" % (k, v) for k, v in request.environ.items()) + ) def subview(request): - return HttpResponse('subview') + return HttpResponse("subview") def subview_calling_view(request): - with urlopen(request.GET['url'] + '/subview/') as response: - return HttpResponse('subview calling view: {}'.format(response.read().decode())) + with urlopen(request.GET["url"] + "/subview/") as response: + return HttpResponse("subview calling view: {}".format(response.read().decode())) def check_model_instance_from_subview(request): - with urlopen(request.GET['url'] + '/create_model_instance/'): + with urlopen(request.GET["url"] + "/create_model_instance/"): pass - with urlopen(request.GET['url'] + '/model_view/') as response: - return HttpResponse('subview calling view: {}'.format(response.read().decode())) + with urlopen(request.GET["url"] + "/model_view/") as response: + return HttpResponse("subview calling view: {}".format(response.read().decode())) @csrf_exempt |