Consider supporting a controlled clock (e.g., similar to async-solipsism) in TestClient

[posita]

It's unfortunate that—even as of this writing—asyncio apparently lacks decent abstractions to allow deterministic advancement of clocks in testing like Twisted has.

I have unit tests that validate functionality that rely on both aiohttp and asyncio.sleep. aiohttp's TestClient allows me to test aiohttp functionality, and async-solipsism affords a loop deterministic clock advancement without affecting wall time. However, I cannot use both at the same time, likely due to limitations in async-solipsism. Consider:

import async_solipsism
import pytest
from aiohttp import web


@pytest.fixture
def event_loop():
    loop = async_solipsism.EventLoop()
    yield loop
    loop.close()


@pytest.fixture
def fake_client(
    event_loop,  # noqa: F811 # pylint: disable=redefined-outer-name
    aiohttp_client,
):
    app = web.Application()
    return event_loop.run_until_complete(aiohttp_client(app))


async def test_integration(
    fake_client,  # noqa: F811 # pylint: disable=redefined-outer-name
):
    resp = await fake_client.post("/hey", json={})
    assert resp.status == 404

This will result in the following error at runtime of tests:

/…/lib/python3.9/asyncio/base_events.py:642: in run_until_complete
    return future.result()
/…/site-packages/aiohttp/pytest_plugin.py:360: in go
    await client.start_server()
/…/site-packages/aiohttp/test_utils.py:268: in start_server
    await self._server.start_server(loop=self._loop)
/…/site-packages/aiohttp/test_utils.py:113: in start_server
    await site.start()
/…/site-packages/aiohttp/web_runner.py:230: in start
    self._server = await loop.create_server(
/…/site-packages/async_solipsism/loop.py:116: in create_server
    return await super().create_server(
/…/lib/python3.9/asyncio/base_events.py:1515: in create_server
    server._start_serving()
/…/lib/python3.9/asyncio/base_events.py:314: in _start_serving
    self._loop._start_serving(
/…/lib/python3.9/asyncio/selector_events.py:150: in _start_serving
    self._add_reader(sock.fileno(), self._accept_connection,
/…/lib/python3.9/asyncio/selector_events.py:261: in _add_reader
    key = self._selector.get_key(fd)
/…/lib/python3.9/selectors.py:191: in get_key
    return mapping[fileobj]
/…/lib/python3.9/selectors.py:72: in __getitem__
    fd = self._selector._fileobj_lookup(fileobj)

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <async_solipsism.selector.Selector object at 0x102dbec40>, fileobj = 17

    def _fileobj_lookup(self, fileobj):
        self._check_closed()
        if isinstance(fileobj, socket.Socket):
            return fileobj.fileno()
        elif isinstance(fileobj, socket.SocketFd):
            return fileobj
        else:
>           raise SolipsismError('Only instances of Socket or SocketFd can be registered')
E           async_solipsism.exceptions.SolipsismError: Only instances of Socket or SocketFd can be registered

Potential solutions

I would love a way to deterministically advance a clock in testing in a way compatible with aiohttp. That could mean refactoring TestClient to work within async-solipsism's limitations. It could mean expanding async-solipsism to allow use with TestClient without modification to the latter. It could mean augmenting aiohttp's internal testing loop to allow more control over the clock. It could mean something else entirely. I'm not sure.

[bmerry]

@posita would you be able to post a complete example that I can run and use for investigation? The code you've posted has missing imports.

[posita]

Ah! Yes. Apologies. I corrected my original post, but see posita/aiohttp-solipsism for the test case in isolation. To run:

git clone https://github.com/posita/aiohttp-solipsism.git
cd aiohttp-solipsism
# mkvirtualenv -a "$( pwd )" aiohttp-solipsism  # or whatever
pip install --editable .
pytest -vv tests

☝️ That should work, but please let me know if it doesn't.

[bmerry]

I still need to read up about pytest-aiohttp (I've used aiohttp before but don't think I've used pytest-aiohttp), but if I delete all the references to async_solipsism from that file the test just hangs. I'd like to start from something that works without async-solipsism before trying to make async-solipsism work.

[Dreamsorcerer]

Seems that aiohttp_client only works directly in the test, or in a fixture only when run using the aiohttp loop fixture. So, this code works for me:

import pytest
from aiohttp import web

@pytest.fixture
def fake_client(loop, aiohttp_client):
    app = web.Application()
    return loop.run_until_complete(aiohttp_client(app))

async def test_integration(fake_client):
    resp = await fake_client.post("/hey", json={})
    assert resp.status == 404

If I load the aiohttp_client without using that loop, then fake_client.post() hangs.

Edit: as mentioned below, the event_loop() fixture must be renamed to override the loop() fixture. aiohttp_client depends on the loop fixture, so you end up trying to use the client with 2 different loops otherwise.

[bmerry]

Give the code below a try, together with the fix-create-server-explicit-socket branch of async-solipsism (it fixes a bug I found while investigating). A new requirement is pytest-mock.

import async_solipsism
import pytest
from aiohttp import web


@pytest.fixture
def loop(mocker):
    mocker.patch(
        "aiohttp.test_utils.get_port_socket",
        lambda host, port: async_solipsism.ListenSocket((host, port)),
    )
    loop = async_solipsism.EventLoop()
    yield loop
    loop.close()


async def test_integration(aiohttp_client):
    app = web.Application()
    client = await aiohttp_client(app, server_kwargs={"port": 80})
    resp = await client.post("/hey", json={})
    assert resp.status == 404

Some things to note:

• The fixture needs to be called loop rather than event_loop because aiohttp's pytest plugin doesn't use the same convention as pytest-asyncio. Once I get everything tidied up I can update the async-solipsism docs for that.
• There is a mock to replace the aiohttp.test_utils function that creates a real socket, so that it creates an async_solipsism fake socket instead. From the aiohttp side it would be useful if that was documented as public API so that tests could reliably depend on it being present to mock.
• The server_kwargs is currently needed because async-solipsism doesn't yet model the normal UNIX sockets behaviour of assigning an unused port when port 0 is requested (it just gives you port 0).

[posita]

Yup! Thanks, @bmerry! I made the above modifications (aiohttp-solipsism@168a544), and it looks like it works!

FWIW, I am not generally a fan of the @pytest.fixture magic. Apologies for not using a more straightforward test setup. Special thanks to @Dreamsorcerer for the assist!

[achimnol]

Maybe could my aiotools.VirtualClock() (example) help you in this case?

[bmerry]

Maybe could my aiotools.VirtualClock() (example) help you in this case?

I did not know about aiotools before. But from a quick look it is just implementing a time-warping clock. My experience with that is that it doesn't work well with networking code (at least on Linux), because the kernel is not infinitely fast: after sending some data into a socketpair, it takes a little time to be available on the other end, and a time-warping clock causes read timeouts to trigger. I created async-solipsism to get around this problem by bypassing the kernel TCP/IP stack and use mock socketpairs which are instantaneous.

[Dreamsorcerer]

FWIW, I am not generally a fan of the @pytest.fixture magic.

Me neither, I just stick to unittest for my own projects.

Is there anything else needed here, or should this be closed now?

[bmerry]

Is there anything else needed here, or should this be closed now?

The example works by mocking an undocumented function inside aiohttp.test_utils (get_port_socket). It would be better it was possible to achieve the goal using public APIs, either by finding another approach, or by having aiohttp document this function.

Does the aiohttp_client fixture allow you to bring your own socket? That would make it possible to create the test server without resorting to mocking at all and without needing to modify aiohttp.

[posita]

@Dreamsorcerer, I think with respect to the narrow question of compatibility with async-solipsism, I'll defer to @bmerry's #5834 (comment).

However, that led me to another, much broader question on what a best practice might be around something like this:

import asyncio
import time
# …
async def test_foo():
    start = time.time()
    await asyncio.sleep(5.0)  # <- has patched loop
    end = time.time()
    assert end - start >= 5.0  # <- will fail

Why is the above important? Because many applications may introduce feature-specific delays and (understandably) want to lean on library interfaces to do that. Consider the following:

import asyncio
import time
from typing import NoReturn

async def poll_state(state) -> NoReturn:
    while True:
        await asyncio.sleep(60.0)
        now = time.time()
        unlocked = state["unlock_time"] < now  # <- problem
        if unlocked:
            state["done"] = True

async def test_poll_state() -> None:
    now = time.time()
    state = {"done": False, "unlock_time": now + 5.0}  # 5 seconds from "now"
    poller = asyncio.create_task(poll_state(state))
    await asyncio.sleep(61.0)  # <- uses mocked loop
    # This will fail because of an advancement mismatch between loop.time() and time.time()
    assert state["done"] == True

☝️ That's not the most testable thing without also mocking time.time, which I really don't like. How are others solving this problem? Are there good patterns (or at least internally consistent tools) to resolve that mismatch?

---

@achimnol, I haven't dug into VirtualClock. I didn't really get a clear sense of how to use it without digging into source code. (The docs are kind of spartan.)

Am I understanding correctly that a minimal application would entail the following?

def test_foo():
    vclock = aiotools.VirtualClock()
    with vclock.patch_loop():
        # do stuff

When I get some more time, I'm happy to try refactoring posita/aiohttp-solipsism to use that and report back here (or maybe file a new issue in achimnol/aiotools if I get stuck). I don't know that we need to keep this bug open for that, though.

[Dreamsorcerer]

The example works by mocking an undocumented function inside aiohttp.test_utils (get_port_socket). It would be better it was possible to achieve the goal using public APIs, either by finding another approach, or by having aiohttp document this function.

I've put this into #5844 to help keep the specific issue clear.

That's not the most testable thing without also mocking time.time, which I really don't like. How are others solving this problem? Are there good patterns (or at least internally consistent tools) to resolve that mismatch?

Seems like a more general problem, my first inclination is that the library involved might consider patching time.time(). If it's for tests, then it might be useful for time.time() to always be based off a set time in relation to the loop, rather than return the actual current timestamp.

I'll close this, but feel free to continue the discussion, and file a new issue if you figure that aiohttp should be providing something else in relation to time.time().

[posita]

Seems like a more general problem, my first inclination is that the library involved might consider patching time.time(). If it's for tests, then it might be useful for time.time() to always be based off a set time in relation to the loop, rather than return the actual current timestamp.

I'm not a fan of that because:

1. It's not known what other things depend on time.time() to continue marching forward without test intervention; and
2. patching time.time() may not be sufficient to handle application code that does from time import time, in which you'd likely have to lean on application authors anyway.

In either case, broadly-applied library patch magic may not be your friend. I think a better solution to this problem is to role model a best practice around a testable time interface. (This may be the purview of something like aiotools or asyncs-solipsism. It could even involve "augmenting" asyncio itself.)

There are probably better solutions out there, but what would be nice is something like asyncio.loop.wall_time() or some such that presented a notion corresponding to epoch time (a la time.time() and may even proxy time.time() as its default implementation), but that could be overridden by test loops to something roughly equivalent to:

class TestLoop(…):
    …
    def wall_time(self) -> float:
        if not hasattr(self, "_start"):
            self._start = time.time() - self.time()  # <- that conceptually belongs in a constructor somewhere
        return self._start + time.time()
    …

More specifically, wall_time() would express the current loop time relative to the loop's start time expressed as seconds since the epoch (i.e., time.time() "compatible").

☝️ If something like that were standard, asyncio application code would "know" to deliberately query asyncio.get_event_loop().wall_time() for time needs. Overriding it in a testing loop would still allow production code to be deterministically testable without requiring patch-ing (and risking side effects). This would obviously require shifting construction of derivative objects. (Calls to datetime.datetime.now(tz=…) would have to be replaced with datetime.datetime.fromtimestamp(asyncio.get_event_loop().wall_time(), tz=…), etc.) There are also gaps around Python's time module's various clocks, but those might be handled separately or left up to the (what I'm guessing is relatively few) application developers who needed that functionality.

One extreme of that trajectory is to re-implement all of time inside asyncio.loop.… somewhere, but I that would probably be overkill.

---

The above is something akin to a more simplistic version of Twisted's twisted.internet.interfaces.IReactorTime as a coarse source of time truth that can be narrowly overridden in tests (e.g., twisted.internet.task.Clock; see also this).

[bmerry]

However, that led me to another, much broader question on what a best practice might be around something like this:

import asyncio
import time
# …
async def test_foo():
    start = time.time()
    await asyncio.sleep(5.0)  # <- has patched loop
    end = time.time()
    assert end - start >= 5.0  # <- will fail

That is poor practice even in the absence of a virtual clock. time.time reports real time, which means it is subject to clock corrections by the sysadmin / NTP / etc. For measuring elapsed time between two events, one should use a monotonic clock.
time.monotonic provides such one such clock, but as far as I know asyncio event loops are not required to match it, just to have a monotonic clock. For asyncio code I recommend using asyncio.get_event_loop().time(), which should always use the same time source as other asyncio functions (including asyncio.sleep), and which will play nicely with async-solipsism (and by the looks of it, aiotools as well).

[posita]

Thanks @bmerry! Agreed that time.time() is probably poor practice, but a separate monotonic clock doesn't really solve the problem either. (I think as you point out) absent a virtual clock that marches in time to the loop-relative clock, practices are bound to be all over the map.

Certainly, where one is in full control of time creation, a monotonic clock is the way to go. However, that's not always the case. Consider the case where one relies on a time value from an external source. (Bitcoin's nLockTime is a good real-world example of this.)

Let's say I'm building an application that waits until my signed Bitcoin transactions are eligible to publish without being considered invalid by neighboring nodes. In practice, my deployment relies on ntp for "close enough" time precision, and I build in a "wait" time of, let's say, 2.0 seconds and implement an exponential back-off into my application just in case. I want to test each of these application behaviors deterministically and without patching. Remember that the source of the time value I'm anchoring on is provided externally (i.e., it's baked into the transaction that cannot be modified without recollecting signatures). I don't know that I have many options in my application to completely avoid an internal clock/source value comparison at some point.

Here's a simplistic implementation of the application code:

import asyncio
import random

async def publish(publisher, transaction):
  locked_until = transaction.nLockTime  # seconds since epoch
  await asyncio.sleep(...)  # <- What goes in here such that I can test this function (e.g., with TestWallet and TestPublisher)?
  sleep_time_sec = 1.0
  while True:
    try:
      await publisher.publish(transaction)
    except PublicationException:
      await asyncio.sleep(sleep_time_sec * random.uniform(0.80, 1.20))
      sleep_time_sec = min(2.0 * sleep_time_sec, 5 * 60.0)  # top out at five minutes
    else:
      break

async def load_and_schedule(wallet, publisher):
  transactions = wallet.load_unpublished_transactions()
  tasks = []
  for transaction in transactions:
    tasks.append(asyncio.create_task(publish(publisher, transaction)))
  return tasks

Twisted standardizes this via a reactor that implements IReactorTime (mentioned above):

import twisted.…
import random

def publish(reactor, publisher, transaction):
  locked_until = transaction.nLockTime  # seconds since epoch
  now = reactor.seconds()  # reactor consolidates the concept of epoch time and loop time into this single interface
  if locked_until + 2.0 < now:
    # Probably not *quite* how one would do this in real life, but it's just to illustrate a point
    reactor.callLater(locked_until + 2.0 - now, publish, reactor, publisher, transaction)
  else:
    # …

# …

That reactor can be substituted at test time without affecting anything about application code, so long as application authors make sure to build time measurements around reactor.seconds().

[posita]

I guess the main thrust of my observation is that, for time-based application code, solely addressing loop.time() advancement is insufficient for many cases. One also has to have access to a non-test mechanism (like a virtual clock) that can be overridden at test time. My argument is that patching library functions is a poor substitute for that mechanism. Absent something built into asyncio itself, there may be an opportunity for authors of loop-advancement tools to provide a more functional primitive. (I also really wish asyncio.loop.time() was called asyncio.loop.clock(), but that ship has long since sailed.)

[posita]

Maybe could my aiotools.VirtualClock() (example) help you in this case?

I did not know about aiotools before. But from a quick look it is just implementing a time-warping clock. My experience with that is that it doesn't work well with networking code (at least on Linux), because the kernel is not infinitely fast: after sending some data into a socketpair, it takes a little time to be available on the other end, and a time-warping clock causes read timeouts to trigger. I created async-solipsism to get around this problem by bypassing the kernel TCP/IP stack and use mock socketpairs which are instantaneous.

@achimnol, I can verify that this is consistent with what I observe with my attempt in posita/aiohttp-solipsism@fd1a6f7. I may have misunderstood the intended use, but when I run my new test, I get:

E           asyncio.exceptions.TimeoutError

Here's how you can run this yourself, if you like:

$ git clone https://github.com/posita/aiohttp-solipsism.git
…
$ cd aiohttp-solipsism
$ mkvirtual env aiohttp-solipsism  # or whatever you prefer
…
$ pip install --upgrade --requirement requirements.txt
…
$ pytest -vv tests/test_also_happy_together.py
…
E           asyncio.exceptions.TimeoutError
…

[achimnol]

The API usage looks right, but my virtual clock has an assumption that the caller is using the standard asyncio event loop.
But async-solipsism provides its own event loop, so it may not work well together... 🤔