Advanced Features

This chapter describes miscellaneous features of asynctest which can be leveraged in specific use cases.

Controlling time

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 instance.

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.

ClockedTestCase is a subclass of TestCase which allows to advance the clock of the loop in a test with the coroutine advance().

This will not affect the wall clock: functions like time.time() or datetime.datetime.now() will return the regular date and time of the system.

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.

Internally, ClockedTestCase will ensure that the loop is executed as if time was passing fast, instead of jumping the clock to the target time.

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 tutorial/clock.py.

Mocking I/O

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 OSError 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.

Mock	spec
FileMock	a file object, implements fileno()
SocketMock	socket.socket
SSLSocketMock	ssl.SSLSocket

We can use asynctest.isfilemock() to differenciate mocks from regular objects.

As of asynctest 0.12, these mocks don’t provide other features, and must be configured to return expected values for calls to methods like read() or recv().

When configured, we still need to force the loop to detect that I/O is possible on these mock files.

This is done with set_read_ready() and set_write_ready().

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. recv_side_effect() makes as if the data is received in several packets, but it has no impact on the high level StreamReader.

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.

It is possible to force the TestCase to use the loop provided by the policy by setting the class attribute use_default_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 asyncio (see 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.

Note

The behavior of asyncio.get_event_loop() changed over time. Explicitly passing the loop is not the recommended practice anymore.