FM / aiohttp

/opt/cloudlinux/venv/lib/python3.11/site-packages/aiohttp/

/opt/cloudlinux/venv/lib/python3.11/site-packages/aiohttp

Name	Size	Mode	Actions
.hash/	-	0755	rm
__pycache__/	-	0755	rm
abc.py	5500	0644	edit dl rm
base_protocol.py	2741	0644	edit dl rm
client.py	47276	0644	edit dl rm
client_exceptions.py	9411	0644	edit dl rm
client_proto.py	8651	0644	edit dl rm
client_reqrep.py	39680	0644	edit dl rm
client_ws.py	11010	0644	edit dl rm
compression_utils.py	5015	0644	edit dl rm
connector.py	52798	0644	edit dl rm
cookiejar.py	14015	0644	edit dl rm
formdata.py	6106	0644	edit dl rm
hdrs.py	4613	0644	edit dl rm
helpers.py	30255	0644	edit dl rm
http.py	1842	0644	edit dl rm
http_exceptions.py	2716	0644	edit dl rm
http_parser.py	35496	0644	edit dl rm
http_websocket.py	26716	0644	edit dl rm
http_writer.py	5933	0644	edit dl rm
locks.py	1136	0644	edit dl rm
log.py	325	0644	edit dl rm
multipart.py	32472	0644	edit dl rm
payload.py	13542	0644	edit dl rm
payload_streamer.py	2087	0644	edit dl rm
py.typed	7	0644	edit dl rm
pytest_plugin.py	11605	0644	edit dl rm
resolver.py	5070	0644	edit dl rm
streams.py	20836	0644	edit dl rm
tcp_helpers.py	961	0644	edit dl rm
test_utils.py	20185	0644	edit dl rm
tracing.py	15132	0644	edit dl rm
typedefs.py	1471	0644	edit dl rm
web.py	19263	0644	edit dl rm
web_app.py	18311	0644	edit dl rm
web_exceptions.py	10360	0644	edit dl rm
web_fileresponse.py	11416	0644	edit dl rm
web_log.py	7801	0644	edit dl rm
web_middlewares.py	4032	0644	edit dl rm
web_protocol.py	23044	0644	edit dl rm
web_request.py	28756	0644	edit dl rm
web_response.py	27729	0644	edit dl rm
web_routedef.py	6132	0644	edit dl rm
web_runner.py	11736	0644	edit dl rm
web_server.py	2587	0644	edit dl rm
web_urldispatcher.py	40057	0644	edit dl rm
web_ws.py	18647	0644	edit dl rm
worker.py	7965	0644	edit dl rm
_cparser.pxd	4318	0644	edit dl rm
_find_header.pxd	68	0644	edit dl rm
_headers.pxi	2007	0644	edit dl rm
_helpers.cpython-311-x86_64-linux-gnu.so	310476	0755	edit dl rm
_helpers.pyi	202	0644	edit dl rm
_helpers.pyx	1049	0644	edit dl rm
_http_parser.cpython-311-x86_64-linux-gnu.so	1889106	0755	edit dl rm
_http_parser.pyx	28058	0644	edit dl rm
_http_writer.cpython-311-x86_64-linux-gnu.so	320845	0755	edit dl rm
_http_writer.pyx	4575	0644	edit dl rm
_websocket.cpython-311-x86_64-linux-gnu.so	187629	0755	edit dl rm
_websocket.pyx	1561	0644	edit dl rm
__init__.py	7762	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/aiohttp/pytest_plugin.py (11605B)

import asyncio
import contextlib
import warnings
from typing import Any, Awaitable, Callable, Dict, Iterator, Optional, Type, Union

import pytest

from aiohttp.helpers import isasyncgenfunction
from aiohttp.web import Application

from .test_utils import (
    BaseTestServer,
    RawTestServer,
    TestClient,
    TestServer,
    loop_context,
    setup_test_loop,
    teardown_test_loop,
    unused_port as _unused_port,
)

try:
    import uvloop
except ImportError:  # pragma: no cover
    uvloop = None  # type: ignore[assignment]

AiohttpClient = Callable[[Union[Application, BaseTestServer]], Awaitable[TestClient]]
AiohttpRawServer = Callable[[Application], Awaitable[RawTestServer]]
AiohttpServer = Callable[[Application], Awaitable[TestServer]]

def pytest_addoption(parser):  # type: ignore[no-untyped-def]
    parser.addoption(
        "--aiohttp-fast",
        action="store_true",
        default=False,
        help="run tests faster by disabling extra checks",
    )
    parser.addoption(
        "--aiohttp-loop",
        action="store",
        default="pyloop",
        help="run tests with specific loop: pyloop, uvloop or all",
    )
    parser.addoption(
        "--aiohttp-enable-loop-debug",
        action="store_true",
        default=False,
        help="enable event loop debug mode",
    )

def pytest_fixture_setup(fixturedef):  # type: ignore[no-untyped-def]
    """Set up pytest fixture.

Allow fixtures to be coroutines. Run coroutine fixtures in an event loop.
    """
    func = fixturedef.func

if isasyncgenfunction(func):
        # async generator fixture
        is_async_gen = True
    elif asyncio.iscoroutinefunction(func):
        # regular async fixture
        is_async_gen = False
    else:
        # not an async fixture, nothing to do
        return

strip_request = False
    if "request" not in fixturedef.argnames:
        fixturedef.argnames += ("request",)
        strip_request = True

def wrapper(*args, **kwargs):  # type: ignore[no-untyped-def]
        request = kwargs["request"]
        if strip_request:
            del kwargs["request"]

# if neither the fixture nor the test use the 'loop' fixture,
        # 'getfixturevalue' will fail because the test is not parameterized
        # (this can be removed someday if 'loop' is no longer parameterized)
        if "loop" not in request.fixturenames:
            raise Exception(
                "Asynchronous fixtures must depend on the 'loop' fixture or "
                "be used in tests depending from it."
            )

_loop = request.getfixturevalue("loop")

if is_async_gen:
            # for async generators, we need to advance the generator once,
            # then advance it again in a finalizer
            gen = func(*args, **kwargs)

def finalizer():  # type: ignore[no-untyped-def]
                try:
                    return _loop.run_until_complete(gen.__anext__())
                except StopAsyncIteration:
                    pass

request.addfinalizer(finalizer)
            return _loop.run_until_complete(gen.__anext__())
        else:
            return _loop.run_until_complete(func(*args, **kwargs))

fixturedef.func = wrapper

@pytest.fixture
def fast(request):  # type: ignore[no-untyped-def]
    """--fast config option"""
    return request.config.getoption("--aiohttp-fast")

@pytest.fixture
def loop_debug(request):  # type: ignore[no-untyped-def]
    """--enable-loop-debug config option"""
    return request.config.getoption("--aiohttp-enable-loop-debug")

@contextlib.contextmanager
def _runtime_warning_context():  # type: ignore[no-untyped-def]
    """Context manager which checks for RuntimeWarnings.

This exists specifically to
    avoid "coroutine 'X' was never awaited" warnings being missed.

If RuntimeWarnings occur in the context a RuntimeError is raised.
    """
    with warnings.catch_warnings(record=True) as _warnings:
        yield
        rw = [
            "{w.filename}:{w.lineno}:{w.message}".format(w=w)
            for w in _warnings
            if w.category == RuntimeWarning
        ]
        if rw:
            raise RuntimeError(
                "{} Runtime Warning{},\n{}".format(
                    len(rw), "" if len(rw) == 1 else "s", "\n".join(rw)
                )
            )

@contextlib.contextmanager
def _passthrough_loop_context(loop, fast=False):  # type: ignore[no-untyped-def]
    """Passthrough loop context.

Sets up and tears down a loop unless one is passed in via the loop
    argument when it's passed straight through.
    """
    if loop:
        # loop already exists, pass it straight through
        yield loop
    else:
        # this shadows loop_context's standard behavior
        loop = setup_test_loop()
        yield loop
        teardown_test_loop(loop, fast=fast)

def pytest_pycollect_makeitem(collector, name, obj):  # type: ignore[no-untyped-def]
    """Fix pytest collecting for coroutines."""
    if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj):
        return list(collector._genfunctions(name, obj))

def pytest_pyfunc_call(pyfuncitem):  # type: ignore[no-untyped-def]
    """Run coroutines in an event loop instead of a normal function call."""
    fast = pyfuncitem.config.getoption("--aiohttp-fast")
    if asyncio.iscoroutinefunction(pyfuncitem.function):
        existing_loop = pyfuncitem.funcargs.get(
            "proactor_loop"
        ) or pyfuncitem.funcargs.get("loop", None)
        with _runtime_warning_context():
            with _passthrough_loop_context(existing_loop, fast=fast) as _loop:
                testargs = {
                    arg: pyfuncitem.funcargs[arg]
                    for arg in pyfuncitem._fixtureinfo.argnames
                }
                _loop.run_until_complete(pyfuncitem.obj(**testargs))

return True

def pytest_generate_tests(metafunc):  # type: ignore[no-untyped-def]
    if "loop_factory" not in metafunc.fixturenames:
        return

loops = metafunc.config.option.aiohttp_loop
    avail_factories: Dict[str, Type[asyncio.AbstractEventLoopPolicy]]
    avail_factories = {"pyloop": asyncio.DefaultEventLoopPolicy}

if uvloop is not None:  # pragma: no cover
        avail_factories["uvloop"] = uvloop.EventLoopPolicy

if loops == "all":
        loops = "pyloop,uvloop?"

factories = {}  # type: ignore[var-annotated]
    for name in loops.split(","):
        required = not name.endswith("?")
        name = name.strip(" ?")
        if name not in avail_factories:  # pragma: no cover
            if required:
                raise ValueError(
                    "Unknown loop '%s', available loops: %s"
                    % (name, list(factories.keys()))
                )
            else:
                continue
        factories[name] = avail_factories[name]
    metafunc.parametrize(
        "loop_factory", list(factories.values()), ids=list(factories.keys())
    )

@pytest.fixture
def loop(loop_factory, fast, loop_debug):  # type: ignore[no-untyped-def]
    """Return an instance of the event loop."""
    policy = loop_factory()
    asyncio.set_event_loop_policy(policy)
    with loop_context(fast=fast) as _loop:
        if loop_debug:
            _loop.set_debug(True)  # pragma: no cover
        asyncio.set_event_loop(_loop)
        yield _loop

@pytest.fixture
def proactor_loop():  # type: ignore[no-untyped-def]
    policy = asyncio.WindowsProactorEventLoopPolicy()  # type: ignore[attr-defined]
    asyncio.set_event_loop_policy(policy)

with loop_context(policy.new_event_loop) as _loop:
        asyncio.set_event_loop(_loop)
        yield _loop

@pytest.fixture
def unused_port(aiohttp_unused_port: Callable[[], int]) -> Callable[[], int]:
    warnings.warn(
        "Deprecated, use aiohttp_unused_port fixture instead",
        DeprecationWarning,
        stacklevel=2,
    )
    return aiohttp_unused_port

@pytest.fixture
def aiohttp_unused_port() -> Callable[[], int]:
    """Return a port that is unused on the current host."""
    return _unused_port

@pytest.fixture
def aiohttp_server(loop: asyncio.AbstractEventLoop) -> Iterator[AiohttpServer]:
    """Factory to create a TestServer instance, given an app.

aiohttp_server(app, **kwargs)
    """
    servers = []

async def go(app, *, port=None, **kwargs):  # type: ignore[no-untyped-def]
        server = TestServer(app, port=port)
        await server.start_server(loop=loop, **kwargs)
        servers.append(server)
        return server

yield go

async def finalize() -> None:
        while servers:
            await servers.pop().close()

loop.run_until_complete(finalize())

@pytest.fixture
def test_server(aiohttp_server):  # type: ignore[no-untyped-def]  # pragma: no cover
    warnings.warn(
        "Deprecated, use aiohttp_server fixture instead",
        DeprecationWarning,
        stacklevel=2,
    )
    return aiohttp_server

@pytest.fixture
def aiohttp_raw_server(loop: asyncio.AbstractEventLoop) -> Iterator[AiohttpRawServer]:
    """Factory to create a RawTestServer instance, given a web handler.

aiohttp_raw_server(handler, **kwargs)
    """
    servers = []

async def go(handler, *, port=None, **kwargs):  # type: ignore[no-untyped-def]
        server = RawTestServer(handler, port=port)
        await server.start_server(loop=loop, **kwargs)
        servers.append(server)
        return server

yield go

async def finalize() -> None:
        while servers:
            await servers.pop().close()

loop.run_until_complete(finalize())

@pytest.fixture
def raw_test_server(  # type: ignore[no-untyped-def]  # pragma: no cover
    aiohttp_raw_server,
):
    warnings.warn(
        "Deprecated, use aiohttp_raw_server fixture instead",
        DeprecationWarning,
        stacklevel=2,
    )
    return aiohttp_raw_server

@pytest.fixture
def aiohttp_client(
    loop: asyncio.AbstractEventLoop,
) -> Iterator[AiohttpClient]:
    """Factory to create a TestClient instance.

aiohttp_client(app, **kwargs)
    aiohttp_client(server, **kwargs)
    aiohttp_client(raw_server, **kwargs)
    """
    clients = []

async def go(
        __param: Union[Application, BaseTestServer],
        *args: Any,
        server_kwargs: Optional[Dict[str, Any]] = None,
        **kwargs: Any
    ) -> TestClient:

if isinstance(__param, Callable) and not isinstance(  # type: ignore[arg-type]
            __param, (Application, BaseTestServer)
        ):
            __param = __param(loop, *args, **kwargs)
            kwargs = {}
        else:
            assert not args, "args should be empty"

if isinstance(__param, Application):
            server_kwargs = server_kwargs or {}
            server = TestServer(__param, loop=loop, **server_kwargs)
            client = TestClient(server, loop=loop, **kwargs)
        elif isinstance(__param, BaseTestServer):
            client = TestClient(__param, loop=loop, **kwargs)
        else:
            raise ValueError("Unknown argument type: %r" % type(__param))

await client.start_server()
        clients.append(client)
        return client

yield go

async def finalize() -> None:
        while clients:
            await clients.pop().close()

loop.run_until_complete(finalize())

@pytest.fixture
def test_client(aiohttp_client):  # type: ignore[no-untyped-def]  # pragma: no cover
    warnings.warn(
        "Deprecated, use aiohttp_client fixture instead",
        DeprecationWarning,
        stacklevel=2,
    )
    return aiohttp_client

cmd: