Async support for qtbot

[vxgmichel]

Hello,

Is there any plan to support some kind of wrapper on top of qtbot in order to turn the wait* methods into async methods that could be called from a qtrio test? That would be very useful in my opinion, e.g:

@pytest.mark.trio(run=qtrio.run)
async def test(aqtbot):
    w = mywidget()
    aqtbot.addWidget(w)
    async with aqtbot.waitSignal(w.mysignal):
        aqtbot.mouseClick(w, QtCore.Qt.LeftButton)

The problem with using the blocking methods directly is that all trio code is blocked while waiting for the corresponding signal. I experimented with it using the following test:

import pytest
import trio
import qtrio

from PyQt5.QtCore import QTimer


@pytest.mark.trio(run=qtrio.run)
async def test(qtbot):

    async with trio.open_nursery() as nursery:

        timer1 = QTimer()
        timer1.setSingleShot(True)
        timer1.setInterval(200)

        timer2 = QTimer()
        timer2.setSingleShot(True)
        timer2.setInterval(200)

        async def aslot():
            await trio.sleep(.2)
            timer2.start()

        timer1.timeout.connect(
            lambda: nursery.start_soon(aslot)
        )
        with qtbot.wait_signal(timer2.timeout):
            timer1.start()
            # await trio.sleep(1)  # Uncomment to fix

The blocking methods are listed below:

• wait
• waitActive
• waitCallback
• waitExposed
• waitForWindowShown
• waitSignal
• waitSignals
• waitUntil

My guess is that there are two ways to approach this:

• Either wrap blocking qtbot calls with some qtrio API, e.g:

  await qtrio.run_blocking_call(qtbot.wait, 1000)

• Or re-implement the methods using the qtrio emission API (e.g enter_emissions_channel)

Also, thanks for the library :)

[altendky]

Welp, aside from a few people that looked at it in the very early days, you are the first that seems to maybe be using QTrio? Welcome. :]

I have wondered whether it makes sense to provide 'pytest-qt support' or just reimplement the features. I haven't taken the time to look into pytest-qt in detail yet. My impression at this point is that a lot of what it provides is the "async" of the blocking functions you refer to. That when you call them it keeps the Qt loop going for you. This is exactly what QTrio makes available directly with await so adding a compatibility layer to put the pytest-qt layer on top isn't obviously the best solution. But maybe.

Aside from an implementation of this functionality, with or without pytest-qt, I'm curious to know what situation you are developing in. Do you have tests that need to work with and without QTrio? If compatibility is needed, would a matching API from another library cut it or does it need to be pytest-qt specifically?

Without claiming these are solutions, here are the first thoughts of related code that exists and might aid thinking about the topic.

• wait
  • await trio.sleep(37/1000)
• waitCallback
  • event = trio.Event(); f(callback=event.set); await event.wait()
  • But sure, this doesn't capture arguments.
• waitSignal

  • qtrio/qtrio/_core.py

    Lines 116 to 145 in 727ba64

    	async def wait_signal(signal: qtrio._util.SignalInstance) -> typing.Tuple[object, ...]:
    	"""Block for the next emission of ``signal`` and return the emitted arguments.
    	Warning:
    	In many cases this can result in a race condition since you are unable to
    	first connect the signal and then wait for it.
    	Args:
    	signal: The signal instance to wait for emission of.
    	Returns:
    	A tuple containing the values emitted by the signal.
    	"""
    	event = trio.Event()
    	result: typing.Tuple[object, ...] = ()
    	def slot(*args: object) -> None:
    	"""Receive and store the emitted arguments and set the event so we can continue.
    	Args:
    	args: The arguments emitted from the signal.
    	"""
    	nonlocal result
    	result = args
    	event.set()
    	with qtrio._qt.connection(signal, slot):
    	await event.wait()
    	return result

  • qtrio/qtrio/_core.py

    Lines 406 to 423 in 727ba64

    	@async_generator.asynccontextmanager
    	async def wait_signal_context(
    	signal: qtrio._util.SignalInstance,
    	) -> typing.AsyncGenerator[None, None]:
    	"""Connect a signal during the context and wait for it on exit. Presently no
    	mechanism is provided for retrieving the emitted arguments.
    	Args:
    	signal: The signal to connect to and wait for.
    	"""
    	event = trio.Event()
    	def slot(*args: object, **kwargs: object) -> None:
    	event.set()
    	with qtrio._qt.connection(signal=signal, slot=slot):
    	yield
    	await event.wait()

There's my first dump. I'll read through your scenario and see what it looks like. Thanks for checking in. :]

[vxgmichel]

Thanks for quick reply!

My impression at this point is that a lot of what it provides is the "async" of the blocking functions you refer to. That when you call them it keeps the Qt loop going for you. This is exactly what QTrio makes available directly with await so adding a compatibility layer to put the pytest-qt layer on top isn't obviously the best solution. But maybe.

Right, that makes sense!

Aside from an implementation of this functionality, with or without pytest-qt, I'm curious to know what situation you are developing in. Do you have tests that need to work with and without QTrio? If compatibility is needed, would a matching API from another library cut it or does it need to be pytest-qt specifically?

We're currently considering a qtrio migration. At the moment we're using separate threads for Qt and each instance of a trio-based user session (the application can host several independent user sessions so it is fine to run each of them in a different trio loop). The communication between threads is done through a job scheduler we developed that lets the qt thread fire jobs that will run within the trio thread and trigger some qt signals when it's done. This works fine but the logic gets quite complicated to maintain when the workflow goes back and forth between qt and trio (error handling is especially tricky). Instead, qtrio would simply let us call some qt within a trio coroutine, and we could rely on standard python flow control (try-except, etc.).

My experimentation with qtrio was almost painless since all that's needed is a re-implementation of the job scheduler on top of a simple trio nursery. Migrating the test suite is bit tougher though since we wrote our own trio/qt test runner and wrapper on top of qtbot. In our case, each test is a trio coroutine that blocks the qt execution until a call to qtbot is performed. Still, the migration should mostly be ok if we replace our current async-wrapper of qtbot by a qtrio-compatible implementation. Something that would probably look like this:

class AsyncQtBot(pytestqt.qtbot.QtBot):

    @asynccontextmanager
    async def waitSignal(self, signal, timeout=5000):
        with trio.fail_after(timeout):
            async with qtrio.wait_signal_context(signal):
                yield

Without claiming these are solutions, here are the first thoughts of related code that exists and might aid thinking about the topic [...]

Thanks for the reference! I didn't know about wait_signal_context, this is going to be very useful.

[altendky]

Well, be wary as it isn't documented and as such isn't super public. :] But, maybe it should be, or something like it.

Anyways, it sounds like just a similar set of functions would satisfy your needs. Do you want to take a pass at it? It could end up here or in pytest-qt itself perhaps. Kind of depends on what it looks like but I suspect there's various sync code that could be reused, or at least ought to be used for reference. But yeah, big picture, this functionality ought to be easily usable with QTrio.