执行内容协商的模拟HTTP服务器

Question

我通常使用Python的http.server来模拟HTTP服务器，但这次我需要服务器来执行内容协商。你知道吗

我只需要它返回一个错误，如果它不能匹配接受头。Python的服务器忽略了头。你知道吗

运行服务器的方法可以是在线程中，也可以是在单独的进程中，比如运行python -m http.server。你知道吗

Answer 1

您可以，仍然使用http.server，但是使用自定义处理程序。。。你知道吗

class MockHandler(http.server.BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'

    def do_GET(self):
        if 'text/html' not in self.headers['Accept']:
            self.send_response(406)
        else:
            self.send_response(200)
        self.end_headers()
        self.wfile.write(b'some data')

class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    pass

server_address = ('localhost', 8888)
server = ThreadedHTTPServer(server_address, MockHandler)
server.daemon_threads = True
thread = threading.Thread(target=server.serve_forever)
thread.daemon = True
thread.start()

（注意：这是一个HTTP1.1服务器，因为现在大多数服务器都是线程/多请求安全的，但是如果您愿意，您可能可以简化……）

下面是一个小测试套件中的完整示例，以及您可能要测试的示例“生产”函数。你知道吗

import http.server
import socketserver
import threading
import unittest
import urllib.request


def example_production_function(accept_header):
    req = urllib.request.Request('http://localhost:8888', headers={'Accept': accept_header});
    try:
        urllib.request.urlopen(req)
        return 'Allowed'
    except urllib.error.HTTPError as e:
        return 'Not allowed'


class TestServer(unittest.TestCase):

    def test(self):

        class MockHandler(http.server.BaseHTTPRequestHandler):
            protocol_version = 'HTTP/1.1'

            def do_GET(self):
                if 'text/html' not in self.headers['Accept']:
                    self.send_response(406)
                else:
                    self.send_response(200)
                self.end_headers()
                self.wfile.write(b'some data')

        class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
            pass

        server_address = ('localhost', 8888)
        server = ThreadedHTTPServer(server_address, MockHandler)
        server.daemon_threads = True
        thread = threading.Thread(target=server.serve_forever)
        thread.daemon = True
        thread.start()

        result = example_production_function('text/html')
        self.assertEqual(result, 'Allowed')

        result = example_production_function('application/json')
        self.assertEqual(result, 'Not allowed')

        server.shutdown()
        server.server_close()


if __name__ == '__main__':
    unittest.main()