Mock Asyncio Subprocess Communicate

AsyncioExample.py

import asyncio
from typing import List


class AsyncioExample:

    async def list_files(self) -> List[str]:
        print("Retrieving list of files")
        try:
            process = await asyncio.create_subprocess_shell(
                " ".join(["ls", "-A"]),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, stderr = await process.communicate()
            if stderr:
                raise stderr
            return stdout.decode('utf-8').split('\n')
        except Exception as err:
            print(err)
            return []

Pipfile

[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[dev-packages]
mock = "*"
pytest-mock = "*"
pytest-asyncio = "*"
asynctest = "*"
pytest = "*"
snapshottest = "*"

[packages]

[requires]
python_version = "3.7"

[pipenv]
allow_prereleases = true

snap_TestAsyncioExample.py

# -*- coding: utf-8 -*-
# snapshottest: v1 - https://goo.gl/zC4yUc
from __future__ import unicode_literals

from snapshottest import Snapshot


snapshots = Snapshot()

snapshots['TestAsyncioExample.test_list_files 1'] = [
    'a.txt',
    'b.txt',
    'c.txt'
]

TestAsyncioExample.py

import asynctest
import mock
import pytest

from src.utils.AsyncioExample import AsyncioExample


class TestAsyncioExample:
    @asynctest.mock.patch("asyncio.create_subprocess_shell")
    @pytest.mark.asyncio
    async def test_list_files(self, mock_async_subproc, snapshot):
        process_mock = mock.Mock()
        future = asynctest.asyncio.Future()
        future.set_result(
            (
                b"""a.txt
b.txt
c.txt""",
                b"",
            )
        )
        attrs = {"communicate.return_value": future}
        process_mock.configure_mock(**attrs)
        mock_async_subproc.return_value = process_mock
        asyncio_example = AsyncioExample()
        output = await asyncio_example.list_files()
        snapshot.assert_match(output)