# 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 replaces the clock of the loop in a test with a manually-controlled one. The clock will only advance when calling `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 asyncio-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 asyncio-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.