🏡 index : quine-global-labs/mirror-cockpit-project-cockpit.git

import asyncio
import os
import subprocess
import sys
from typing import AsyncGenerator

import pytest
import pytest_asyncio

from cockpit._vendor.systemd_ctypes import EventLoopPolicy

collect_ignore: list[str] = []
if sys.version_info < (3, 12):
    # Not used from cockpit-bridge and only intended for Python 3.12+
    collect_ignore.append("ws")


def any_subprocesses() -> bool:
    # Make sure we don't leak subprocesses
    try:
        os.waitid(os.P_ALL, -1, os.WEXITED | os.WNOHANG | os.WNOWAIT)
    except ChildProcessError:
        return False  # good !
    else:
        return True  # at least one process (or zombie) still waitable


# some issues:
#   - even sourcing this will break old python versions in ways we don't care to fix
#   - even sourcing this requires aiohttp, which isn't available everywhere
#   - there are general issues running under tox which need investigation
#   - this also requires a bots/ checkout which we don't have in distros
if "NO_QUNIT" in os.environ:
    collect_ignore.append("test_browser.py")
else:
    pytest_plugins = "js_coverage"


@pytest.fixture
# AbstractEventLoopPolicy is deprecated in 3.14 and will be removed in 3.16
def event_loop_policy() -> asyncio.AbstractEventLoopPolicy:  # type: ignore[name-defined]
    return EventLoopPolicy()


@pytest_asyncio.fixture(autouse=True)
async def _check_settled() -> AsyncGenerator[None, None]:
    yield

    current_task = asyncio.current_task()
    assert current_task is not None
    event_loop = asyncio.get_running_loop()

    # Let all tasks and subprocesses run to completion
    for _ in range(200):
        all_tasks = asyncio.all_tasks(loop=event_loop) - {current_task}
        if not (all_tasks or any_subprocesses()):
            break
        await asyncio.sleep(0.005)

    # No tasks left
    assert asyncio.all_tasks(loop=event_loop) == {current_task}

    # No subprocesses left
    if any_subprocesses():
        # Bad news.  Show some helpful output.
        subprocess.run(['ps', 'f', f'--pid={os.getpid()}', f'--ppid={os.getpid()}'])
        # clear it out for the sake of the other tests
        subprocess.run(['pkill', '-9', '-P', f'{os.getpid()}'])
        try:
            for _ in range(100):  # zombie vacuum
                os.wait()
        except ChildProcessError:
            pass

        pytest.fail('Some subprocesses still running!')