无法为 pytest 停止龙卷风
Cannot stop Tornado for pytest
我目前遇到问题,无法测试我们的 tornado 网络服务器。
首先:Tornado 运行在单独的线程上运行。
我的 Wrapper Class 定义如下:
class TornadoProvider():
def Stop(self):
tornado.ioloop.IOLoop.instance().add_callback(tornado.ioloop.IOLoop.instance().stop)
def Start(self):
asyncio.set_event_loop(asyncio.new_event_loop())
application = tornado.web.Application([
(r'/', VibraSocket),
(r'/api/message', MessageHandler),
(r'/api/settings', SettingsHandler),
], debug=True)
application.listen(8080)
tornado.ioloop.IOLoop.instance().start()
我可以在我们的应用程序中使用这个 class:
self.TornadoServer = TornadoProvider.TornadoProvider()
threading.Thread(target=self.TornadoServer.Start).start()
并用以下方法杀死它:
self.TornadoServer.Stop()
现在的主要问题是,当我 运行 pytest 时,测试用例 运行 通过并成功,但之后 pytest 陷入了无限循环。出现此问题是因为 Tornado IO 循环未停止。我确实尝试了所有方法(甚至在断言之前手动调用 Stop-Method),但没有任何效果。下面是我测试的代码 class.
class TestTornadoApiCalls:
TornadoServer = None
def setup_class(self):
self.TornadoServer = TornadoProvider.TornadoProvider()
threading.Thread(target=self.TornadoServer.Start).start()
time.sleep(0.5)
def test_send_message(self):
message = Message(MessageTypes.Notification, datetime.now(), "Test")
endpoint = "..."
response = requests.post(url = endpoint, data = message.ToJson())
assert response.text == "OK"
def test_get_settings(self):
endpoint = "..."
response = requests.get(url = endpoint)
with open('./Files/settings.json', 'r') as file:
data = json.loads(file.read())
assert response.text == str(data).replace("'", "\"")
def teardown_class(self):
self.TornadoServer.Stop()
也许有人可以引导我走向正确的方向。
默认情况下,即使父线程死亡,Python 的子线程也会继续 运行。但是给Thread
参数daemon=None
,没有父线程的子线程就会死掉。或者你应该 join()
在 teadown_class()
.
IOLoop.instance()
returns IOLoop
的线程特定实例。这意味着当你在 Start
中调用它时,它 returns 你为线程创建的实例,但在 Stop
中,它 returns 主线程的 IOLoop 实例(它你实际上从来没有 运行)。您必须保存在 Start 中创建的 IOLoop 对象,并在 Stop 中引用 that 对象而不是主线程中 IOLoop.instance()
returns 的对象。
我目前遇到问题,无法测试我们的 tornado 网络服务器。 首先:Tornado 运行在单独的线程上运行。
我的 Wrapper Class 定义如下:
class TornadoProvider():
def Stop(self):
tornado.ioloop.IOLoop.instance().add_callback(tornado.ioloop.IOLoop.instance().stop)
def Start(self):
asyncio.set_event_loop(asyncio.new_event_loop())
application = tornado.web.Application([
(r'/', VibraSocket),
(r'/api/message', MessageHandler),
(r'/api/settings', SettingsHandler),
], debug=True)
application.listen(8080)
tornado.ioloop.IOLoop.instance().start()
我可以在我们的应用程序中使用这个 class:
self.TornadoServer = TornadoProvider.TornadoProvider()
threading.Thread(target=self.TornadoServer.Start).start()
并用以下方法杀死它:
self.TornadoServer.Stop()
现在的主要问题是,当我 运行 pytest 时,测试用例 运行 通过并成功,但之后 pytest 陷入了无限循环。出现此问题是因为 Tornado IO 循环未停止。我确实尝试了所有方法(甚至在断言之前手动调用 Stop-Method),但没有任何效果。下面是我测试的代码 class.
class TestTornadoApiCalls:
TornadoServer = None
def setup_class(self):
self.TornadoServer = TornadoProvider.TornadoProvider()
threading.Thread(target=self.TornadoServer.Start).start()
time.sleep(0.5)
def test_send_message(self):
message = Message(MessageTypes.Notification, datetime.now(), "Test")
endpoint = "..."
response = requests.post(url = endpoint, data = message.ToJson())
assert response.text == "OK"
def test_get_settings(self):
endpoint = "..."
response = requests.get(url = endpoint)
with open('./Files/settings.json', 'r') as file:
data = json.loads(file.read())
assert response.text == str(data).replace("'", "\"")
def teardown_class(self):
self.TornadoServer.Stop()
也许有人可以引导我走向正确的方向。
默认情况下,即使父线程死亡,Python 的子线程也会继续 运行。但是给Thread
参数daemon=None
,没有父线程的子线程就会死掉。或者你应该 join()
在 teadown_class()
.
IOLoop.instance()
returns IOLoop
的线程特定实例。这意味着当你在 Start
中调用它时,它 returns 你为线程创建的实例,但在 Stop
中,它 returns 主线程的 IOLoop 实例(它你实际上从来没有 运行)。您必须保存在 Start 中创建的 IOLoop 对象,并在 Stop 中引用 that 对象而不是主线程中 IOLoop.instance()
returns 的对象。