(self)
| 227 | self.assertEqual(response_body["body"], b"Echo!") |
| 228 | |
| 229 | async def test_request_too_big_request_error(self): |
| 230 | # Track request_finished signal. |
| 231 | signal_handler = SignalHandler() |
| 232 | request_finished.connect(signal_handler) |
| 233 | self.addCleanup(request_finished.disconnect, signal_handler) |
| 234 | |
| 235 | # Request class that always fails creation with RequestDataTooBig. |
| 236 | class TestASGIRequest(ASGIRequest): |
| 237 | |
| 238 | def __init__(self, scope, body_file): |
| 239 | super().__init__(scope, body_file) |
| 240 | raise RequestDataTooBig() |
| 241 | |
| 242 | # Handler to use the custom request class. |
| 243 | class TestASGIHandler(ASGIHandler): |
| 244 | request_class = TestASGIRequest |
| 245 | |
| 246 | application = TestASGIHandler() |
| 247 | scope = self.async_request_factory._base_scope(path="/not-important/") |
| 248 | communicator = ApplicationCommunicator(application, scope) |
| 249 | |
| 250 | # Initiate request. |
| 251 | await communicator.send_input({"type": "http.request"}) |
| 252 | # Give response.close() time to finish. |
| 253 | await communicator.wait() |
| 254 | |
| 255 | self.assertEqual(len(signal_handler.calls), 1) |
| 256 | self.assertNotEqual( |
| 257 | signal_handler.calls[0]["thread"], threading.current_thread() |
| 258 | ) |
| 259 | |
| 260 | async def test_meta_not_modified_with_repeat_headers(self): |
| 261 | scope = self.async_request_factory._base_scope(path="/", http_version="2.0") |
nothing calls this directly
no test coverage detected