How to unit test socketio AsyncServer deployed with Sanic using unittest.TestCase?

[nsmkhn]

Hi @miguelgrinberg

I would like to unit test my simple service implemented using socketio.AsyncServer & Sanic.

class EventService:
    def __init__(self, port=8000, listener_timeout=5):
        self.port = port
        self.event_listener = EventListener()
        self.listener_timeout = listener_timeout
        self.sio = socketio.AsyncServer(async_mode='sanic', logger=True, engineio_logger=True)
        self.app = Sanic('event_service')
        self.sio.attach(self.app)
        self.__setup_sio_event_handlers()
        self.__register_app_listeners()

    def run(self):
        self.app.run(port=self.port)

    async def __gather_and_send_events(self):
        while True:
            await self.sio.sleep(self.listener_timeout)
            for event in self.event_listener.get_events():
                for topic in event.topics:
                    await self.sio.emit('event', event.data, room=topic)

    def __setup_sio_event_handlers(self):
        @self.sio.event
        async def join(sid, msg):
            self.sio.enter_room(sid, msg['room'])
            await self.sio.emit('message', 'Entered room: ' + msg['room'], room=sid)

        @self.sio.event
        async def leave(sid, msg):
            self.sio.leave_room(sid, msg['room'])
            await self.sio.emit('message', 'Left room: ' + msg['room'], room=sid)

    def __register_app_listeners(self):
        @self.app.listener('after_server_start')
        def __start_event_listener_task(app, loop):
            self.sio.start_background_task(self.__gather_and_send_events)

if __name__ == '__main__':
    service = EventService()
    service.run()

I'm struggling with understanding how to launch app instance so I can connect to it from several testcases. I'd like to have the following tests working:

class TestEventService(TestCase):
    def setUp(self):
        self.service = EventService()
        self.client = MockSIOClient()

    def test_service_can_run(self):
        self.assertTrue(self.service.app.is_running)

    @async_test
    async def test_service_is_reachable_by_clients(self):
        await self.client.sio.connect('http://localhost:8000')
        self.assertTrue(self.client.is_connected)
        self.client.sio.disconnect()

    @async_test
    async def test_client_can_join_room(self):
        await self.sio.emit('join', {'room': 'some_room'})
        self.assertTrue(self.client.is_in_room('some_room'))

    @async_test
    async def test_client_can_leave_room(self):
        await self.sio.emit('leave', {'room': 'some_room'})
        self.assertFalse(self.client.is_in_room('some_room'))

    @async_test
    async def test_client_receives_events(self):
        w = EventWriter()
        w.write_event(Event(topics['some_room'], number=101, hash_str='xoxo'))
        # EventService pickups event and sends to client
        self.assertEqual(len(self.client.events), 1)

I would really appreciate your advice on how to make this work. Thanks in advance!

[miguelgrinberg]

I have no experience unit testing with Sanic, so I cannot help you there. For the Socket.IO server you have two options, you can start a real server and then connect to it with a real client, or else you can just call your Socket.IO handlers directly and make sure they do what they need to do (mocking any emits or other things that need to be suppressed during tests).

Related: #465

[nsmkhn]

Thank you for response. In order to connect to my socketio server with a real client I need to start the sanic app which will block the further execution until it's down. I'll try to put it in a separate thread, maybe it helps. Will see how it goes.

Regarding second approach you've mentioned: I think I understood what you mean will also give it a try.

[miguelgrinberg]

Why does Sanic block? You should be able to start the server, and while the server is running start an async task for your test(s). It may be tricky to orchestrate these two things to start on the same loop, but it should be possible.

You should also be able to run the Sanic app in a thread, with one async loop, and the tests in another thread, with a second loop.

[nsmkhn]

@miguelgrinberg thank you, I was able to achieve what I wanted using python multiprocessing. For future readers here is how it can be done:

class TestEventService(TestCase):
    service_process = None

    @staticmethod
    def start_event_service():
        EventService(port=EVENT_SERVICE_PORT).run()

    @classmethod
    def setUpClass(cls):
        TestEventService.service_process = Process(target=TestEventService.start_event_service)
        TestEventService.service_process.start()
        time.sleep(WAIT_SERVICE_ACTION_COMPLETE_TIMEOUT)

    @classmethod
    def tearDownClass(cls):
        TestEventService.service_process.terminate()
        asyncio.get_event_loop().close()

    def setUp(self):
        self.client = MockSIOClient()
        self.loop = asyncio.get_event_loop()

    def test_service_is_reachable_by_clients(self):
        self.loop.run_until_complete(self.client.sio.connect('http://localhost:{}'.format(EVENT_SERVICE_PORT)))
        self.assertTrue(self.client.is_connected)

    def test_client_can_join_and_leave_room(self):
        self.loop.run_until_complete(self.client.sio.connect('http://localhost:{}'.format(EVENT_SERVICE_PORT)))
        self.loop.run_until_complete(self.client.sio.emit('join', {'room': SAMPLE_TOPIC}))
        self.loop.run_until_complete(asyncio.sleep(WAIT_SERVICE_ACTION_COMPLETE_TIMEOUT))
        self.assertTrue(SAMPLE_TOPIC in self.client.rooms)
        self.loop.run_until_complete(self.client.sio.emit('leave', {'room': SAMPLE_TOPIC}))
        self.loop.run_until_complete(asyncio.sleep(WAIT_SERVICE_ACTION_COMPLETE_TIMEOUT))
        self.assertTrue(SAMPLE_TOPIC not in self.client.rooms)

    def test_service_sends_events_correctly(self):
        self.loop.run_until_complete(self.client.sio.connect('http://localhost:{}'.format(EVENT_SERVICE_PORT)))
        self.loop.run_until_complete(self.client.sio.emit('join', {'room': SAMPLE_TOPIC}))
        EventWriter().write_event(SAMPLE_EVENT)
        EventWriter().write_event(Event(topics=['other_topic', SAMPLE_TOPIC], number=101, hash_str='xoxo'))
        self.loop.run_until_complete(asyncio.sleep(WAIT_SERVICE_ACTION_COMPLETE_TIMEOUT))
        self.assertEqual(len(self.client.events), 2)

@miguelgrinberg I'd appreciate your comments regarding if this could be done simpler or any other notes