Test asynchronous code with PyTest (translation)
When preparing the material for the course , we occasionally come across interesting articles that I would like to share with you!
By Stefan Scherfke “Testing (asyncio) coroutines with pytest”
')
PyTest is an excellent package for testing in Python, and for a long time one of my favorite packages in general. It greatly facilitates the writing of tests and has ample opportunities to compile reports on unsuccessful tests.
However, at the time of version 2.7, it is less effective in testing (asyncio) routines. Therefore, do not try to test them in this way:
In this method, many shortcomings and excesses. The only interesting lines are those containing the statements yield from and assert.
Each test case must have its own event cycle, which is completed correctly, regardless of the success of the test.
Applying yield in the manner described above will not work, pytest will decide that our test returns new test cases.
Therefore, it is necessary to create a separate subroutine in which the actual test is contained, and for its execution an event loop is started.
Tests will be cleaner if done like this:
Pytest has a flexible plugin system, which makes implementation of this behavior possible. But, unfortunately, most of the necessary hooks are documented poorly or not at all, therefore, finding out how to perform them is problematic.
A local pop-up plugin is created simply because it is a bit simpler than creating a “real” external plugin. Pytest finds in each test directory a file called conftest.py and applies fixtures and hooks, implemented in it, to all the tests in this directory.
Let's start by writing a fixture that creates a new event loop for each test case and closes it correctly at the end of the test:
Before each test, pytest executes a fixture loop to the first yield statement. What returns returns is passed as an argument to the loop (that is, the loop) of our test case. When the test is completed (successfully or not), pytest terminates the loop fixture, closing it correctly. In the same way, you can write a test fixture that creates a socket and closes it after each test (the fixture of the socket may depend on the cycle fixture in the same way as in our example. Cool, isn't it?)
But the end is still far away. We need to teach pytest to execute our test routines. To do this, change how the asyncio subroutines are assembled (they should be assembled as normal test functions, and not as test generators) and how they are executed (using loop.run_until_complete ()):
This plugin works in pytest version 2.4 and higher. I checked its performance with versions 2.6 and 2.7.
And everything would be fine, but soon after the publication of this solution in Stack Overflow, the plugin PyTest-Asyncio appeared, but Stefan was not upset at all, but did a detailed analysis of this plugin.
In my first article, I showed how pytest helps write quality tests. Fixtures allow you to create a clean event loop for each test case, and thanks to the plugin system you can write test functions, which are actually asyncio coroutines.
But while work was being done on this material, Tin Tvrtkowitz created the pytest-asyncio plugin.
In short, it allows you to do this:
Instead:
Using pytest-asyncio visually improves the test (and this is not the limit of the plugin’s possibilities!).
When I was working on aiomas, I encountered additional requirements that were not so easy to accomplish.
A little bit about what is aiomas itself. He adds three layers of abstraction asyncio transports:
The simplest example of how the channel layer works:
Need a clean event loop for each test.
This can be done using the event_loop fixture, which is in pytest-asyncio.
Each test must be run with every possible transport (TCP socket, Unix domain socket, ...).
Theoretically, this can be solved with the help of the pytest.mark.parametrize () decorator (but not in our case, how further it will become clear).
Each test needs a client coroutine. Ideally, the test itself.
The pytest.mark.asyncio decorator in pytest-asyncio does the job.
Each test requires a server with a custom callback for client connections. At the end of the test, the servers should be turned off regardless of the test result.
It seems that the coroutine can solve this problem, but each server needs a specific callback to manage client connections. Which complicates problem solving. I do not want to receive the “Address already in use” error if one of the tests fails. The fixture unused_tcp_port in pytest-asyncio to the rescue.
I do not want to constantly use loop.run_until_complete ().
And the pytest.mark.asyncio decorator solves the problem.
To summarize what remains to be solved: each test needs two fixtures (one for the event loop and one more for the address type), but I want to combine them into one. You need to create a fixture to configure the server, but how to do it?
You can wrap the loop and the type of address in the fixture. Let's call it ctx (abbreviated from test context). Due to the fixture parameters, it is easy to create a separate address for each type.
This allows you to write tests like this:
It is already working well, and each test using fixture ctx, runs once for each type of address.
However, two problems remain:
Let's see how you can solve these problems.
The first problem is solved by using the getfuncargvalue () method belonging to the request object, which our fixture receives. This method can manually call its function:
To solve the second problem, you can extend the Context class, which is passed to each test. Add the Context.start_server (client_handler) method, which can be called directly from the tests. And also add the final teardown to our ctx fixture, which will close the server after the end. In addition, you can create several functions for shortcuts.
The test case becomes noticeably shorter, easier to read and more reliable thanks to this additional functionality:
The ctx fixture (and Context class) is certainly not the shortest thing I ever wrote, but it helped rid my tests of about 200 lines of sample code.
Elegant solutions and reliable code!
By Stefan Scherfke “Testing (asyncio) coroutines with pytest”
')
PyTest is an excellent package for testing in Python, and for a long time one of my favorite packages in general. It greatly facilitates the writing of tests and has ample opportunities to compile reports on unsuccessful tests.
However, at the time of version 2.7, it is less effective in testing (asyncio) routines. Therefore, do not try to test them in this way:
# tests/test_coros.py import asyncio def test_coro(): loop = asyncio.get_event_loop() @asyncio.coroutine def do_test(): yield from asyncio.sleep(0.1, loop=loop) assert 0 # onoes! loop.run_until_complete(do_test()) In this method, many shortcomings and excesses. The only interesting lines are those containing the statements yield from and assert.
Each test case must have its own event cycle, which is completed correctly, regardless of the success of the test.
Applying yield in the manner described above will not work, pytest will decide that our test returns new test cases.
Therefore, it is necessary to create a separate subroutine in which the actual test is contained, and for its execution an event loop is started.
Tests will be cleaner if done like this:
# tests/test_coros.py @asyncio.coroutine def test_coro(loop): yield from asyncio.sleep(0.1, loop=loop) assert 0 Pytest has a flexible plugin system, which makes implementation of this behavior possible. But, unfortunately, most of the necessary hooks are documented poorly or not at all, therefore, finding out how to perform them is problematic.
A local pop-up plugin is created simply because it is a bit simpler than creating a “real” external plugin. Pytest finds in each test directory a file called conftest.py and applies fixtures and hooks, implemented in it, to all the tests in this directory.
Let's start by writing a fixture that creates a new event loop for each test case and closes it correctly at the end of the test:
# tests/conftest.py import asyncio import pytest @pytest.yield_fixture def loop(): # loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) yield loop # loop.close() # tests/test_coros.py def test_coro(loop): @asyncio.coroutine def do_test(): yield from asyncio.sleep(0.1, loop=loop) assert 0 # onoes! loop.run_until_complete(do_test()) Before each test, pytest executes a fixture loop to the first yield statement. What returns returns is passed as an argument to the loop (that is, the loop) of our test case. When the test is completed (successfully or not), pytest terminates the loop fixture, closing it correctly. In the same way, you can write a test fixture that creates a socket and closes it after each test (the fixture of the socket may depend on the cycle fixture in the same way as in our example. Cool, isn't it?)
But the end is still far away. We need to teach pytest to execute our test routines. To do this, change how the asyncio subroutines are assembled (they should be assembled as normal test functions, and not as test generators) and how they are executed (using loop.run_until_complete ()):
# tests/conftest.py def pytest_pycollect_makeitem(collector, name, obj): """ asyncio , .” if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj): # # , # , "pytest.mark.parametrize()". return list(collector._genfunctions(name, obj)) # else: # None, pytest' “obj” def pytest_pyfunc_call(pyfuncitem): """ ``pyfuncitem.obj`` - asyncio , """ testfunction = pyfuncitem.obj if not asyncio.iscoroutinefunction(testfunction): # None, . Pytest # return # : funcargs = pyfuncitem.funcargs # argnames = pyfuncitem._fixtureinfo.argnames # testargs = {arg: funcargs[arg] for arg in argnames} # - ( !) coro = testfunction(**testargs) # loop = testargs['loop'] if loop in testargs else asyncio.get_event_loop() loop.run_until_complete(coro) return True # pytest', This plugin works in pytest version 2.4 and higher. I checked its performance with versions 2.6 and 2.7.
And everything would be fine, but soon after the publication of this solution in Stack Overflow, the plugin PyTest-Asyncio appeared, but Stefan was not upset at all, but did a detailed analysis of this plugin.
Advanced asynchronous testing
In my first article, I showed how pytest helps write quality tests. Fixtures allow you to create a clean event loop for each test case, and thanks to the plugin system you can write test functions, which are actually asyncio coroutines.
But while work was being done on this material, Tin Tvrtkowitz created the pytest-asyncio plugin.
In short, it allows you to do this:
import asyncio import time import pytest @pytest.mark.asyncio async def test_coro(event_loop): before = time.monotonic() await asyncio.sleep(0.1, loop=event_loop) after = time.monotonic() assert after - before >= 0.1 Instead:
import asyncio import time def test_coro(): loop = asyncio.new_event_loop() try: asyncio.set_event_loop(loop) before = time.monotonic() loop.run_until_complete(asyncio.sleep(0.1, loop=loop)) after = time.monotonic() assert after - before >= 0.1 finally: loop.close() Using pytest-asyncio visually improves the test (and this is not the limit of the plugin’s possibilities!).
When I was working on aiomas, I encountered additional requirements that were not so easy to accomplish.
A little bit about what is aiomas itself. He adds three layers of abstraction asyncio transports:
- The channel layer allows you to send messages in JSON or MsgPack format using the “request-response” scheme. It uses a custom protocol that works with different types of transports: TCP sockets, Unix domain sockets and custom transport - local queue.
- The RPC layer creates a Remote Procedure Call system above the channel layer.
- The agent layer (for multi-agent systems) contains even more details related to the network, and allows you to write classes that call methods of other classes through a network connection.
The simplest example of how the channel layer works:
import aiomas async def handle_client(channel): """ """ req = await channel.recv() print(req.content) await req.reply('cya') await channel.close() async def client(): """ : """ channel = await aiomas.channel.open_connection(('localhost', 5555)) rep = await channel.send('ohai') print(rep) await channel.close() server = aiomas.run(aiomas.channel.start_server( ('localhost', 5555), handle_client)) aiomas.run(client()) server.close() aiomas.run(server.wait_closed()) Test requirements
Need a clean event loop for each test.
This can be done using the event_loop fixture, which is in pytest-asyncio.
Each test must be run with every possible transport (TCP socket, Unix domain socket, ...).
Theoretically, this can be solved with the help of the pytest.mark.parametrize () decorator (but not in our case, how further it will become clear).
Each test needs a client coroutine. Ideally, the test itself.
The pytest.mark.asyncio decorator in pytest-asyncio does the job.
Each test requires a server with a custom callback for client connections. At the end of the test, the servers should be turned off regardless of the test result.
It seems that the coroutine can solve this problem, but each server needs a specific callback to manage client connections. Which complicates problem solving. I do not want to receive the “Address already in use” error if one of the tests fails. The fixture unused_tcp_port in pytest-asyncio to the rescue.
I do not want to constantly use loop.run_until_complete ().
And the pytest.mark.asyncio decorator solves the problem.
To summarize what remains to be solved: each test needs two fixtures (one for the event loop and one more for the address type), but I want to combine them into one. You need to create a fixture to configure the server, but how to do it?
First approach
You can wrap the loop and the type of address in the fixture. Let's call it ctx (abbreviated from test context). Due to the fixture parameters, it is easy to create a separate address for each type.
import tempfile import py import pytest class Context: def __init__(self, loop, addr): self.loop = loop self.addr = addr @pytest.fixture(params=['tcp', 'unix']) def ctx(request, event_loop, unused_tcp_port, short_tmpdir): """ TCP Unix.""" addr_type = request.param if addr_type == 'tcp': addr = ('127.0.0.1', unused_tcp_port) elif addr_type == 'unix': addr = short_tmpdir.join('sock').strpath else: raise RuntimeError('Unknown addr type: %s' % addr_type) ctx = Context(event_loop, addr) return ctx @pytest.yield_fixture() def short_tmpdir(): """ Unix. , pytest' tmpdir, """ with tempfile.TemporaryDirectory() as tdir: yield py.path.local(tdir) This allows you to write tests like this:
import aiomas @pytest.mark.asyncio async def test_channel(ctx): results = [] async def handle_client(channel): req = await channel.recv() results.append(req.content) await req.reply('cya') await channel.close() server = await aiomas.channel.start_server(ctx.addr, handle_client) try: channel = await aiomas.channel.open_connection(ctx.addr) rep = await channel.send('ohai') results.append(rep) await channel.close() finally: server.close() await server.wait_closed() assert results == ['ohai', 'cya'] It is already working well, and each test using fixture ctx, runs once for each type of address.
However, two problems remain:
- Fixture always requires an unused TCP port + temporary directory - despite the fact that we need only one of the two.
- Setting up a server (and closing it) includes a certain amount of code that will be identical for all tests, and therefore should be included in the fixture. But it will not work directly, because each server needs a test-dependent callback (this can be seen in the line where we create the server server = await ...). But without a server fixture, there is no teardown for it ...
Let's see how you can solve these problems.
Second approach
The first problem is solved by using the getfuncargvalue () method belonging to the request object, which our fixture receives. This method can manually call its function:
@pytest.fixture(params=['tcp', 'unix']) def ctx(request, event_loop): """ TCP Unix.""" addr_type = request.param if addr_type == 'tcp': port = request.getfuncargvalue('unused_tcp_port') addr = ('127.0.0.1', port) elif addr_type == 'unix': tmpdir = request.getfuncargvalue('short_tmpdir') addr = tmpdir.join('sock').strpath else: raise RuntimeError('Unknown addr type: %s' % addr_type) ctx = Context(event_loop, addr) return ctx To solve the second problem, you can extend the Context class, which is passed to each test. Add the Context.start_server (client_handler) method, which can be called directly from the tests. And also add the final teardown to our ctx fixture, which will close the server after the end. In addition, you can create several functions for shortcuts.
import asyncio import tempfile import py import pytest class Context: def __init__(self, loop, addr): self.loop = loop self.addr = addr self.server = None async def connect(self, **kwargs): """ "self.addr".""" return (await aiomas.channel.open_connection( self.addr, loop=self.loop, **kwargs)) async def start_server(self, handle_client, **kwargs): """ *handle_client*, "self.addr".""" self.server = await aiomas.channel.start_server( self.addr, handle_client, loop=self.loop, **kwargs) async def start_server_and_connect(self, handle_client, server_kwargs=None, client_kwargs=None): """ :: await ctx.start_server(...) channel = await ctx.connect()" """ if server_kwargs is None: server_kwargs = {} if client_kwargs is None: client_kwargs = {} await self.start_server(handle_client, **server_kwargs) return (await self.connect(**client_kwargs)) async def close_server(self): """ .""" if self.server is not None: server, self.server = self.server, None server.close() await server.wait_closed() @pytest.yield_fixture(params=['tcp', 'unix']) def ctx(request, event_loop): """ TCP Unix.""" addr_type = request.param if addr_type == 'tcp': port = request.getfuncargvalue('unused_tcp_port') addr = ('127.0.0.1', port) elif addr_type == 'unix': tmpdir = request.getfuncargvalue('short_tmpdir') addr = tmpdir.join('sock').strpath else: raise RuntimeError('Unknown addr type: %s' % addr_type) ctx = Context(event_loop, addr) yield ctx # : aiomas.run(ctx.close_server()) aiomas.run(asyncio.gather(*asyncio.Task.all_tasks(event_loop), return_exceptions=True)) The test case becomes noticeably shorter, easier to read and more reliable thanks to this additional functionality:
import aiomas @pytest.mark.asyncio async def test_channel(ctx): results = [] async def handle_client(channel): req = await channel.recv() results.append(req.content) await req.reply('cya') await channel.close() channel = await ctx.start_server_and_connect(handle_client) rep = await channel.send('ohai') results.append(rep) await channel.close() assert results == ['ohai', 'cya'] The ctx fixture (and Context class) is certainly not the shortest thing I ever wrote, but it helped rid my tests of about 200 lines of sample code.
The end
Elegant solutions and reliable code!
Source: https://habr.com/ru/post/337108/
All Articles