tests.py 5.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150
  1. # -*- coding: utf-8 -*-
  2. from __future__ import unicode_literals
  3. from django.core.handlers.wsgi import WSGIHandler, WSGIRequest
  4. from django.core.signals import request_started, request_finished
  5. from django.db import close_old_connections, connection
  6. from django.test import RequestFactory, TestCase, TransactionTestCase
  7. from django.test import override_settings
  8. from django.utils.encoding import force_str
  9. from django.utils import six
  10. class HandlerTests(TestCase):
  11. def setUp(self):
  12. request_started.disconnect(close_old_connections)
  13. def tearDown(self):
  14. request_started.connect(close_old_connections)
  15. # Mangle settings so the handler will fail
  16. @override_settings(MIDDLEWARE_CLASSES=42)
  17. def test_lock_safety(self):
  18. """
  19. Tests for bug #11193 (errors inside middleware shouldn't leave
  20. the initLock locked).
  21. """
  22. # Try running the handler, it will fail in load_middleware
  23. handler = WSGIHandler()
  24. self.assertEqual(handler.initLock.locked(), False)
  25. with self.assertRaises(Exception):
  26. handler(None, None)
  27. self.assertEqual(handler.initLock.locked(), False)
  28. def test_bad_path_info(self):
  29. """Tests for bug #15672 ('request' referenced before assignment)"""
  30. environ = RequestFactory().get('/').environ
  31. environ['PATH_INFO'] = b'\xed' if six.PY2 else '\xed'
  32. handler = WSGIHandler()
  33. response = handler(environ, lambda *a, **k: None)
  34. self.assertEqual(response.status_code, 400)
  35. def test_non_ascii_query_string(self):
  36. """
  37. Test that non-ASCII query strings are properly decoded (#20530, #22996).
  38. """
  39. environ = RequestFactory().get('/').environ
  40. raw_query_strings = [
  41. b'want=caf%C3%A9', # This is the proper way to encode 'café'
  42. b'want=caf\xc3\xa9', # UA forgot to quote bytes
  43. b'want=caf%E9', # UA quoted, but not in UTF-8
  44. b'want=caf\xe9', # UA forgot to convert Latin-1 to UTF-8 and to quote (typical of MSIE)
  45. ]
  46. got = []
  47. for raw_query_string in raw_query_strings:
  48. if six.PY3:
  49. # Simulate http.server.BaseHTTPRequestHandler.parse_request handling of raw request
  50. environ['QUERY_STRING'] = str(raw_query_string, 'iso-8859-1')
  51. else:
  52. environ['QUERY_STRING'] = raw_query_string
  53. request = WSGIRequest(environ)
  54. got.append(request.GET['want'])
  55. if six.PY2:
  56. self.assertListEqual(got, ['café', 'café', 'café', 'café'])
  57. else:
  58. # On Python 3, %E9 is converted to the unicode replacement character by parse_qsl
  59. self.assertListEqual(got, ['café', 'café', 'caf\ufffd', 'café'])
  60. def test_non_ascii_cookie(self):
  61. """Test that non-ASCII cookies set in JavaScript are properly decoded (#20557)."""
  62. environ = RequestFactory().get('/').environ
  63. raw_cookie = 'want="café"'
  64. if six.PY3:
  65. raw_cookie = raw_cookie.encode('utf-8').decode('iso-8859-1')
  66. environ['HTTP_COOKIE'] = raw_cookie
  67. request = WSGIRequest(environ)
  68. # If would be nicer if request.COOKIES returned unicode values.
  69. # However the current cookie parser doesn't do this and fixing it is
  70. # much more work than fixing #20557. Feel free to remove force_str()!
  71. self.assertEqual(request.COOKIES['want'], force_str("café"))
  72. @override_settings(ROOT_URLCONF='handlers.urls')
  73. class TransactionsPerRequestTests(TransactionTestCase):
  74. available_apps = []
  75. def test_no_transaction(self):
  76. response = self.client.get('/in_transaction/')
  77. self.assertContains(response, 'False')
  78. def test_auto_transaction(self):
  79. old_atomic_requests = connection.settings_dict['ATOMIC_REQUESTS']
  80. try:
  81. connection.settings_dict['ATOMIC_REQUESTS'] = True
  82. response = self.client.get('/in_transaction/')
  83. finally:
  84. connection.settings_dict['ATOMIC_REQUESTS'] = old_atomic_requests
  85. self.assertContains(response, 'True')
  86. def test_no_auto_transaction(self):
  87. old_atomic_requests = connection.settings_dict['ATOMIC_REQUESTS']
  88. try:
  89. connection.settings_dict['ATOMIC_REQUESTS'] = True
  90. response = self.client.get('/not_in_transaction/')
  91. finally:
  92. connection.settings_dict['ATOMIC_REQUESTS'] = old_atomic_requests
  93. self.assertContains(response, 'False')
  94. @override_settings(ROOT_URLCONF='handlers.urls')
  95. class SignalsTests(TestCase):
  96. def setUp(self):
  97. self.signals = []
  98. self.signaled_environ = None
  99. request_started.connect(self.register_started)
  100. request_finished.connect(self.register_finished)
  101. def tearDown(self):
  102. request_started.disconnect(self.register_started)
  103. request_finished.disconnect(self.register_finished)
  104. def register_started(self, **kwargs):
  105. self.signals.append('started')
  106. self.signaled_environ = kwargs.get('environ')
  107. def register_finished(self, **kwargs):
  108. self.signals.append('finished')
  109. def test_request_signals(self):
  110. response = self.client.get('/regular/')
  111. self.assertEqual(self.signals, ['started', 'finished'])
  112. self.assertEqual(response.content, b"regular content")
  113. self.assertEqual(self.signaled_environ, response.wsgi_request.environ)
  114. def test_request_signals_streaming_response(self):
  115. response = self.client.get('/streaming/')
  116. self.assertEqual(self.signals, ['started'])
  117. self.assertEqual(b''.join(response.streaming_content), b"streaming content")
  118. self.assertEqual(self.signals, ['started', 'finished'])
  119. @override_settings(ROOT_URLCONF='handlers.urls')
  120. class HandlerSuspiciousOpsTest(TestCase):
  121. def test_suspiciousop_in_view_returns_400(self):
  122. response = self.client.get('/suspicious/')
  123. self.assertEqual(response.status_code, 400)