如何为Python socketserver请求处理程序编写单元测试？

Question

我该如何为一个socketserver的请求处理器编写单元测试呢？我尝试使用以下代码，但遇到了一些问题：

import socketserver, unittest, threading

class TestServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    allow_reuse_address = True

class MyRequestHandlerTest(unittest.TestCase):
    def setUp(self):
        self.server = TestServer((HOST, PORT), MyRequestHandler)
        self.server_thread = threading.Thread(target=self.server.serve_forever)
        self.client = socket.create_connection((HOST, PORT))
        #self.server_thread.setDaemon(True)
        self.server_thread.start()

    def tearDown(self):
        self.client.close()
        self.server.shutdown()

如果我尝试使用多个测试函数，像这样：

    def test_foo(self):
        self.client.send(b'foo\n')
        result = self.client.recv(1024)
        self.assertEqual(b'FOO', result)

    def test_bar(self):
        self.client.send(b'bar\n')
        result = self.client.recv(1024)
        self.assertEqual(b'BAR', result)

我在这一行

socket.error: [Errno 98] Address already in use

遇到了错误，虽然我已经告诉测试服务器要allow_reuse_address。

编辑：为了简化问题，我删除了原本发布的两个错误中的一个。

Answer 1

我试过了，创建了一个请求处理器之后，这个方法对我有效：

def tearDown(self): 
    self.client.close() 
    self.server.shutdown()
    self.server.server_close()

可能是socketserver里的shutdown有个bug，它自己没有调用server_close。