This chapter describes miscellaneous features of
asynctest which can be
leveraged in specific use cases.
Tests running calls to
asyncio.sleep() will take as long as the sum of
all these calls. These calls are frequent when testing for timeouts, for
In many cases, this will add a useless delay to the execution of the test suite, and encourage us to deactivate or ignore these tests.
class TestAdvanceTime(asynctest.ClockedTestCase): async def test_advance_time(self): base_loop_time = self.loop.time() base_wall_time = time.time() await self.advance(10) self.assertEqual(base_loop_time + 10, self.loop.time()) self.assertTrue(is_time_around(base_wall_time))
This example is pretty self-explanatory: we verified that the clock of the loop advanced as expected, without awaiting 10 actual seconds and changing the time of the wall clock.
ClockedTestCase will ensure that the loop is
executed as if time was passing fast, instead of jumping the clock to the
class TestWithClockAndCallbacks(asynctest.ClockedTestCase): results = None def runs_at(self, expected_time): self.results.append(is_time_around(expected_time, self.loop)) @asynctest.fail_on(active_handles=True) async def test_callbacks_executed_when_expected(self): self.results =  base_time = self.loop.time() self.loop.call_later(1, self.runs_at, base_time + 1) self.loop.call_at(base_time + 7, self.runs_at, base_time + 7) # This shows that the callback didn't run yet self.assertEqual(0, len(self.results)) await self.advance(10) # This shows that the callbacks ran... self.assertEqual(2, len(self.results)) # ...when expected self.assertTrue(all(self.results))
This example schedules function calls to be executed later by the loop.
Each call will verify that it runs at the expected time.
@fail_on(active_handles=True) ensures that the callbacks have been executed
when the test finishes.
The source code of
is_time_around() can be found in the example file
Testing libraries or functions dealing with low-level IO objects may be complex: these objects are outside of our control, since they are owned by the kernel. It can be impossible to exactly predict their behavior and simulate edge-cases, such as the ones happening in a real-world scenario in a large network.
Even worse, using mocks in place of files will often raise
because these obhjects are not compatible with the features of the system used
by the loop.
asynctest provides special mocks which can be used in place of actual
file-like objects. They are supported by the loop provided by
TestCase if the loop uses a standard implementation with a
selector (Window’s Proactor loop or uvloop are not supported).
These mocks are configured with a spec matching common file-like objects.
||a file object, implements
We can use
asynctest.isfilemock() to differenciate mocks from regular
asynctest 0.12, these mocks don’t provide other features, and must
be configured to return expected values for calls to methods like
When configured, we still need to force the loop to detect that I/O is possible on these mock files.
class TestMockASocket(asynctest.TestCase): async def test_read_and_write_from_socket(self): socket_mock = asynctest.SocketMock() socket_mock.type = socket.SOCK_STREAM recv_data = iter(( b"some data read", b"some other", b" ...and the last", )) recv_buffer = bytearray() def recv_side_effect(max_bytes): nonlocal recv_buffer if not recv_buffer: try: recv_buffer.extend(next(recv_data)) asynctest.set_read_ready(socket_mock, self.loop) except StopIteration: # nothing left pass data = recv_buffer[:max_bytes] recv_buffer = recv_buffer[max_bytes:] if recv_buffer: # Some more data to read asynctest.set_read_ready(socket_mock, self.loop) return data def send_side_effect(data): asynctest.set_read_ready(socket_mock, self.loop) return len(data) socket_mock.recv.side_effect = recv_side_effect socket_mock.send.side_effect = send_side_effect reader, writer = await asyncio.open_connection(sock=socket_mock) writer.write(b"a request?") self.assertEqual(b"some", await reader.read(4)) self.assertEqual(b" data read", await reader.read(10)) self.assertEqual(b"some other ...and the last", await reader.read())
In this example, we configure a socket mock to simulate a simple
request-response scenario with a TCP (stream) socket. Some data is available to
read on the socket once a request has been written.
makes as if the data is received in several packets, but it has no impact on
the high level
It’s common that while a read operation blocks until data is available, a write is often successful. Thus, we didn’t bother simulating the case where the congestion control would block the write operation.
Testing with event loop policies¶
Advanced users may not be able to use the loop provided by
TestCase because they use a customized event loop policy (see
Policies). It is often the case when using an alternative
implementation (like uvloop) or if the
tests are integrated within a framework hidding the scheduling and management
of the loop.
Conversely, authors of libraries may not want to assume which loop they should
use and let users explicitly pass the loop as argument to a function call. For
instance, most of the high-level functions of
Streams, for instance) allow the caller to specify the loop
to use if it needs this kind of flexibility.
forbid_get_event_loop forbids the use of
asyncio.get_event_loop(). An exception is raised if the method is
called while a test is running. It helps developers to ensure they don’t rely
on the default loop this their library.
The behavior of
asyncio.get_event_loop() changed over time.
Explicitly passing the loop is not the recommended practice anymore.