net/netservices2/testserver.py

60     def do_GET(self, write_response=True):
61         authorized, extra_headers = self._authorize()
65         encoding, response_body = self._build_response_body()
67         status_code = extract_desired_status_code_from_path(self.path, 200)
68         self.send_response(status_code)
70             self.send_header('Location', '/')
75             self.send_header('Content-Type', 'text/plain')
76             self.send_header('Content-Length', str(len(response_body)))
78                 self.send_header('Content-Encoding', encoding)
81             self.send_header(header_name, header_value)
82         self.end_headers()
85             self.wfile.write(response_body)
87     def do_HEAD(self):
88         self.do_GET(False)
90     def do_POST(self):
91         authorized, extra_headers = self._authorize()
95         encoding, response_body = self._build_response_body()
96         self.send_response(
97             extract_desired_status_code_from_path(self.path, 200))
98         self.send_header('Content-Type', 'text/plain')
99         self.send_header('Content-Length', str(len(response_body)))
101             self.send_header('Content-Encoding', encoding)
103             self.send_header(header_name, header_value)
105         self.end_headers()
106         self.wfile.write(response_body)
108     def do_DELETE(self):
109         self._not_supported()
111     def do_PATCH(self):
112         self._not_supported()
114     def do_OPTIONS(self):
115         self._not_supported()
117     def send_response(self, code, message=None):
118         self.log_request(code)
119         self.send_response_only(code, message)
120         self.send_header('Server', 'Test HTTP Server for Haiku')
121         self.send_header('Date', 'Sun, 09 Feb 2020 19:32:42 GMT')
123     def _build_response_body(self):
136             for e in self.headers.get('Accept-Encoding', '').split(',')
149             'Path: {}\r\n\r\n'.format(self.path).encode('utf-8'))
152         for header in self.headers:
153             for header_value in self.headers.get_all(header):
171                         self.headers.get('Content-Type', 'text/plain'))
180         content_length = int(self.headers.get('Content-Length', 0))
186             body_bytes = self.rfile.read(content_length).decode('utf-8')
196     def _not_supported(self):
197         self.send_response(405, '{} not supported'.format(self.command))
198         self.end_headers()
199         self.wfile.write(
200             '{} not supported\r\n'.format(self.command).encode('utf-8'))
202     def _authorize(self):
210         match = AUTH_PATH_RE.match(self.path)
219             return self._handle_basic_auth(
223             return self._handle_digest_auth(
230     def _handle_basic_auth(self, expected_username, expected_password):
231         authorization = self.headers.get('Authorization', None)
248             self.send_response(401, 'Not authorized')
249             self.send_header('Www-Authenticate', 'Basic realm="Fake Realm"')
250             self.end_headers()
255     def _handle_digest_auth(self, expected_username, expected_password):
265         authorization = self.headers.get('Authorization', None)
276                 self.command,
277                 self.path,
285             self.send_response(401, 'Not authorized')
286             self.send_header(
294             self.send_header('Set-Cookie', 'stale_after=never; Path=/')
295             self.send_header('Set-Cookie', 'fake=fake_value; Path=/')
296             self.end_headers()
306     def write(self, bytes):
310     def get_bytes(self):
315     def __init__(self):
316         self.buf = io.BytesIO()
318     def write(self, bytes):
319         self.buf.write(bytes)
321     def get_bytes(self):
322         return self.buf.getvalue()
326     def __init__(self):
327         self.buf = io.BytesIO()
328         self.compressor = gzip.GzipFile(
331             fileobj=self.buf)
333     def write(self, bytes):
334         self.compressor.write(bytes)
336     def get_bytes(self):
337         self.compressor.close()
338         return self.buf.getvalue()
342     def __init__(self):
343         self.raw = RawResponseBodyBuilder()
345     def write(self, bytes):
346         self.raw.write(bytes)
348     def get_bytes(self):
349         return zlib.compress(self.raw.get_bytes())
520         help='If set, a self-signed TLS certificate, key and CA will be'