Блог

Как я могу написать модульные тесты для обработчика запросов 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'foon')
        result = self.client.recv(1024)
        self.assertEqual(b'FOO', result)

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

Я получаю сообщение об ошибке

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

в строке self.server = ... , хотя я дал указание тестовому серверу allow_reuse_address .

Редактировать: удалена одна из двух ошибок, которые я изначально опубликовал, чтобы упростить вопрос.

У меня это сработало (после создания RequestHandler) с:

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

Возможно, ошибка в socketserver shutdown заключается в том, что он не вызывает server_close сам себя.