Demo from lighting talks Jul 21, 2020

pyproject.toml

[tool.black]
line-length=79
target-version=["py38"]

[tool.isort]
line_length = 79
use_parentheses = true
balanced_wrapping = true
include_trailing_comma = true
multi_line_output = 3
known_third_party=["pytest", "yarl", "aiohttp", "aioredis", "envparse"]
known_first_party=["aiohttp_cache"]

[tool.coverage.run]
branch = true
source = ["tests"]

[tool.coverage.paths]
source = ["tests/"]

[tool.coverage.report]
fail_under = 70 # TODO should be 95%
skip_covered = true
show_missing = true

[tool.pytest.ini_options]
minversion = "6.0.0rc1"
addopts = "-vv -s --no-cov-on-fail --tb=native"
testpaths = "tests/"
log_cli = true
log_cli_level = "DEBUG"
log_cli_format = "| %(asctime)s | %(name)s | %(levelname)s | %(filename)s | %(message)s"

[tool.poetry]
name = "asyncio-subprocess-shell-demo"
version = "0.1"
description = "A cache system for aiohttp server"
authors = ["zhukovgreen <iam@zhukovgreen.pro>"]
readme = "README.md"
license = "BSD"

[tool.poetry.dependencies]
python = "^3.8"
colorama = "^0.4.3"


[tool.poetry.dev-dependencies]
coverage = {extras = ["toml"], version = "^5.0.3"}
pytest = {version = "^6.0.0-rc.1", allow-prereleases = true}
pytest-cov = "^2.8"
pytest-asyncio = "^0.14.0"


[build-system]
requires = ["poetry>=1.0.0."]
build-backend = "poetry.masonry.api"

test_demo.py

import asyncio
import logging
import sys

import pytest

from colorama import Fore, init

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logger = logging.getLogger(__name__)


async def run(cmd, color: Fore):
    logger.info(color + f"Starting execution of %r" + Fore.RESET, cmd)
    await asyncio.sleep(2)
    proc = await asyncio.create_subprocess_shell(
        cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    stdout, stderr = await proc.communicate()

    logger.info(
        color + f"[{cmd!r} exited with {proc.returncode}]" + Fore.RESET
    )
    if stdout:
        logger.info(color + f"[stdout]\n{stdout.decode()}" + Fore.RESET)
    if stderr:
        logger.warning(color + f"[stderr]\n{stderr.decode()}" + Fore.RESET)


async def daemon_script():
    while True:
        logger.info(Fore.LIGHTMAGENTA_EX + "Doing some job!" + Fore.RESET)
        await asyncio.sleep(1)


@pytest.mark.asyncio
async def test_run():
    init(autoreset=True)
    daemon = asyncio.Task(daemon_script())
    results = asyncio.gather(
        run("ls -al", Fore.RED),
        run("sleep 1; echo 'Finished sleeping'", Fore.BLUE),
        run("ssdsd", Fore.CYAN),
    )
    await results
    daemon.cancel()

Just run pytest on a file. Before that, you have to install dependencies via poetry install. Or do just

pip install pytest==6.0.0-rc.1 pytest-asyncio