Cannot stop Tornado for pytest

Question

I currently have the problem, that I cannot test our tornado webserver. First of all: Tornado is running on a separate thread.

My Wrapper Class is defined as follows:

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()

I can use this class in our application with:

self.TornadoServer = TornadoProvider.TornadoProvider()
threading.Thread(target=self.TornadoServer.Start).start()

and kill it with:

self.TornadoServer.Stop()

The main problem now is, that when I run pytest, the testcases run through and succeed, but afterwards pytest is stuck in an endless loop. This issue occurs, because the Tornado IO Loop is not stopped. I literally tried everything (even manually call the Stop-Method before the assertions) but it has no effect. Below is the code of my test 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()

Maybe someone can lead me into the right direction.

Answer 1

IOLoop.instance() returns a thread-specific instance of IOLoop. That means when you call it in Start, it returns the instance you created for the thread, but in Stop, it returns the IOLoop instance for the main thread (which you never actually run). You must save the IOLoop object created in Start, and in Stop refer to that object instead of the one that IOLoop.instance() returns in the main thread.