| 41 | |
| 42 | |
| 43 | class TestServerRequest(http.Request): |
| 44 | __test__ = False |
| 45 | channel: "TestServerHTTPChannel" |
| 46 | post_body: Optional[bytes] = None |
| 47 | |
| 48 | def process(self) -> None: |
| 49 | server = self.channel.factory.server_instance |
| 50 | if self.content: |
| 51 | self.post_body = self.content.read() |
| 52 | self.content.seek(0, 0) |
| 53 | else: |
| 54 | self.post_body = None |
| 55 | uri = urlparse(self.uri.decode()) |
| 56 | path = uri.path |
| 57 | |
| 58 | request_subscriber = server.request_subscribers.get(path) |
| 59 | if request_subscriber: |
| 60 | request_subscriber._loop.call_soon_threadsafe(request_subscriber.set_result, self) |
| 61 | server.request_subscribers.pop(path) |
| 62 | |
| 63 | if server.auth.get(path): |
| 64 | authorization_header = self.requestHeaders.getRawHeaders("authorization") |
| 65 | creds_correct = False |
| 66 | if authorization_header: |
| 67 | creds_correct = server.auth.get(path) == ( |
| 68 | self.getUser().decode(), |
| 69 | self.getPassword().decode(), |
| 70 | ) |
| 71 | if not creds_correct: |
| 72 | self.setHeader(b"www-authenticate", 'Basic realm="Secure Area"') |
| 73 | self.setResponseCode(HTTPStatus.UNAUTHORIZED) |
| 74 | self.finish() |
| 75 | return |
| 76 | if server.csp.get(path): |
| 77 | self.setHeader(b"Content-Security-Policy", server.csp[path]) |
| 78 | if server.routes.get(path): |
| 79 | server.routes[path](self) |
| 80 | return |
| 81 | file_content = None |
| 82 | try: |
| 83 | file_content = (server.static_path / path[1:]).read_bytes() |
| 84 | content_type = mimetypes.guess_type(path)[0] |
| 85 | if content_type and content_type.startswith("text/"): |
| 86 | content_type += "; charset=utf-8" |
| 87 | self.setHeader(b"Content-Type", content_type) |
| 88 | self.setHeader(b"Cache-Control", "no-cache, no-store") |
| 89 | if path in server.gzip_routes: |
| 90 | self.setHeader("Content-Encoding", "gzip") |
| 91 | self.write(gzip.compress(file_content)) |
| 92 | else: |
| 93 | self.setHeader(b"Content-Length", str(len(file_content))) |
| 94 | self.write(file_content) |
| 95 | self.setResponseCode(HTTPStatus.OK) |
| 96 | except (FileNotFoundError, IsADirectoryError, PermissionError): |
| 97 | self.setResponseCode(HTTPStatus.NOT_FOUND) |
| 98 | self.finish() |
| 99 | |
| 100 |
nothing calls this directly
no outgoing calls
no test coverage detected