bluesky / ophyd-async

Hardware abstraction for bluesky written using asyncio
https://blueskyproject.io/ophyd-async
BSD 3-Clause "New" or "Revised" License
12 stars 26 forks source link

Make `put_mock` an `AsyncMock` #539

Closed DominicOram closed 2 months ago

DominicOram commented 3 months ago

As a developer I was expecting to be able to do the following:


async def my_async_function():
    await ...

callback_on_mock_put(my_mocked_signal, my_mocked_function)

However, because put_mock is a Mock not an AsyncMock I cannot. Instead I have had to go around the mock and do:

@AsyncStatus.wrap
async def my_async_function(value: float, wait=None, timeout=None):
    await ...

my_mocked_signal.set = my_async_function
noemifrisina commented 2 months ago

Adding an example from the PMAC device which has this issue:

@AsyncStatus.wrap
async def set(self, value: int, wait=True, timeout=None):
    prog_str = f"&2b{value}r"
    assert isinstance(timeout, SupportsFloat) or (
        timeout is None
    ), f"ProgramRunner does not support calculating timeout itself, {timeout}"
    await self.signal.set(prog_str, wait=wait)
    # First wait for signal to go to 1, then wait for the scan to finish.
    await wait_for_value(
        self.status,
        ScanState.RUNNING,
        timeout=DEFAULT_TIMEOUT,
    )
    await wait_for_value(self.status, ScanState.DONE, timeout)

In the test, callback_on_mock_put won't take an async function so the workaround is:

async def go_high_then_low():
    set_mock_value(fake_pmac.scanstatus, 1)
    await asyncio.sleep(0.01)
    set_mock_value(fake_pmac.scanstatus, 0)

prog_num = 10
callback_on_mock_put(
    fake_pmac.pmac_string,
    lambda *args, **kwargs: asyncio.create_task(go_high_then_low()),  # type: ignore
)
coretl commented 2 months ago

Would someone from MX like to work on this? Or should we put it on our backlog?

DominicOram commented 2 months ago

If you're happy for us to do it we can, I think it's like a one line change.

coretl commented 2 months ago

Yes please