Testing a python pipeline | Dagger | Page 1

scarlet lava Nov 9, 2023, 1:29 PM

#

Hello all!

I am trying to mock a coroutine, so I can generate some simple unittests for my pipeline. However, this doesn't work as I exepct.

I have following function which is part of the dagger pipeline:

async def unittests(runner):
    return await runner.with_exec(["poetry", "run", "pytest", "tests/unittests"]).stdout()  # (1)

Which I want to test:

@pytest.mark.asyncio
async def test_pipeline():
    await unittests(mock_runner)
    assert mock_runner.with_exec.assert_called_with(["poetry", "run", "pytest", "tests/unittests"])

The problem is, that I cannot build a mock, which provides the .stdout() and therefore the test fails.

await mock_runner.with_exec().stdout()
Traceback (most recent call last):
  File "<string>", line 1, in <module>
AttributeError: 'coroutine' object has no attribute 'stdout'

Any idea how the mock needs to look like?

Of course, that didn't work ...

mock_runner = mock.AsyncMock()
mock_runner.with_exec = mock.AsyncMock()
mock_runner.with_exec.stdout = mock.AsyncMock()

Using mock.MagicMock works as I expect, but does not work within the async/await statement.

jade magnet Nov 9, 2023, 4:12 PM

#

hey! not a python dev myself, maybe @prime jasper has some ideas 🙏

prime jasper Nov 9, 2023, 5:55 PM

#

Hi! Only stdout is a coroutine function if you notice. with_exec is a normal function.

#

mock.AsyncMock subclasses mock.MagicMock, so you just need MagicMock until stdout where you replace with AsyncMock.

scarlet lava Nov 10, 2023, 7:12 AM

#

I tried that, but that is not working within the await statement part.

mock_runner = mock.MagicMock()
mock_runner.with_exec = mock.MagicMock()
mock_runner.with_exec.stdout = mock.AsyncMock()

>>> mock_runner.with_exec().stdout()
<MagicMock name='mock.with_exec().stdout()' id='140724470491664'>
>>> await mock_runner.with_exec().stdout()
Traceback (most recent call last):
  File "<string>", line 1, in <module>
TypeError: object MagicMock can't be used in 'await' expression

So, I am back at the beginning :/

scarlet lava Nov 10, 2023, 8:17 AM

#

I found one workaround.

mock_runner = mock.MagicMock()
mock_runner.with_exec = mock.MagicMock(return_value=mock_runner)
mock_runner.stdout = mock.AsyncMock()

If I let the MagicMock() for with_exec return the "base" mock, than I can also add the AsyncMock() and call it with the await as I want.

>>> await mock_runner.with_exec().stdout()
<AsyncMock name='mock.stdout()' id='140598632986080'>

To be honest, I am not 100% sure, why the first version I tried didn't work.

prime jasper Nov 10, 2023, 12:02 PM

#

This is what I had in mind:

async def under_test(ctr):
    return await ctr.with_exec(["poetry", "run", "pytest"]).stdout()


@pytest.mark.anyio()
async def test_pipeline(mocker):
    mock_ctr = mocker.MagicMock()
    mock_ctr.stdout = mocker.AsyncMock()
    mock_ctr.with_exec.return_value = mock_ctr

    await under_test(mock_ctr)

    mock_ctr.with_exec.assert_called_once_with(["poetry", "run", "pytest"])

It makes sense because you're creating a mock object with an async stdout method and a with_exec that returns that same object, i.e., self.

prime jasper Nov 10, 2023, 12:19 PM

#

That's a more faithful representation of what actually happens compared to your first attempt which doesn't preserve the chaining. The problem in that example is that you're just setting an async method in mock_runner.with_exec, but with_exec needs to return an object with the async stdout and it will default to a new mock, as per the documentation:

Mocks are callable and create attributes as new mocks when you access them [...] return_value: The value returned when the mock is called. By default this is a new Mock (created on first access).
https://docs.python.org/3/library/unittest.mock.html#unittest.mock.Mock

#

So using return_value is not a workaround, but this is the simplest representation:

mock_ctr = mocker.MagicMock()
mock_ctr.stdout = mocker.AsyncMock()
mock_ctr.with_exec.return_value = mock_ctr

#Testing a python pipeline