vinissimus / async-asgi-testclient Goto Github PK

https://2.python-requests.org//en/master/user/quickstart/#cookies

This package currently treats lifespan protocol as mandatory - if the application raises an exception in a lifespan.startup message, it treats the testclient as failed.

The ASGI spec states:

If an exception is raised when calling the application callable with a lifespan.startup message or a scope with type lifespan, the server must continue but not send any lifespan events.
This allows for compatibility with applications that do not support the lifespan protocol. If you want to log an error that occurs during lifespan startup and prevent the server from starting, then send back lifespan.startup.failed instead.

So to test correctly, the TestClient should really allow an ASGI application to raise an exception, and if so then continue without sending any further lifespan messages, including on aexit.

Allow using a tuple for files argument to pass few files with the same name.

This is required to test these endpoints https://fastapi.tiangolo.com/tutorial/request-files/#multiple-file-uploads

Example for requests module: https://stackoverflow.com/a/20769921

Any chance this could be released with a different, more permissive license?

Both of the frameworks you suggest for use with this library (Starlette and Quart) make use of more permissive open source licenses (BSD for Starlette; MIT for Quart).

Given how much activity there is in the ASGI ecosystem right now (and also the extent to which other projects all have permissive licenses), I would be hesitant to start building around a less established dependency with a non-permissive license -- my fear would be that a different project with a more compatible license would end up with more wide-spread adoption in the long term.

(I would totally understand if you prefer to keep the code GPL licensed, just figured it was worth asking.)

I am testing a FastAPI application that actually cares about the request hostname and port. I find async-asgi-testclient much better for our needs than Starlette's TestClient or httpx, however, right now simulating requests that have different hostnames and ports is quite hard.

It would be really cool if instead of just a path, I could pass in a full URL including hostname and port, and these will be taken into account when constructing the request (e.g. with the host header).

I could work around this in my tests by subclassing TestClient and overriding open and websocket_connect, but it would be nice if this would have been a built-in option (potentially, also allowing to set a default base_url, which is a common option in other tests clients).

For reference, here is my overriding code:

class TestClient(BaseTestClient):

    base_url: Optional[str] = None

    async def open(self, path: str, **kwargs: Any):
        path, kwargs = self._fix_args(path, kwargs)
        return await super().open(path, **kwargs)

    def websocket_connect(self, path: str, *args, **kwargs):
        path, kwargs = self._fix_args(path, kwargs)
        if "scheme" in kwargs:
            del kwargs["scheme"]  # TODO: deal with `wss://` connections somehow? - this is a separate issue...
        return super().websocket_connect(path, *args, **kwargs)

    def _fix_args(
        self, path: str, kwargs: Dict[str, Any]
    ) -> Tuple[str, Dict[str, Any]]:
        path, scheme, hostname = self._parse_path_or_url(path)
        headers = kwargs.get("headers", {})
        if hostname and not headers.get("host"):
            headers.update({"host": hostname})
            kwargs["headers"] = headers
        if scheme:
            kwargs["scheme"] = scheme

        return path, kwargs

    def _parse_path_or_url(self, path_or_url: str) -> Tuple[str, str, Optional[str]]:
        if self.base_url and "://" not in path_or_url:
            path_or_url = urljoin(self.base_url, path_or_url)

        if "://" not in path_or_url:
            return path_or_url, "https", None

        parts = urlsplit(path_or_url)
        scheme = parts.scheme
        hostname = parts.hostname
        if parts.port:
            hostname += ":" + str(parts.port)
        path = urlunsplit(("", "", parts.path, parts.query, parts.fragment))
        return path, scheme, hostname

First off, great library! Thanks for putting it out here. However, I'm a having bit of trouble getting ariadne support.

The following:

test.py

from async_asgi_testclient import TestClient
import pytest
from src.core.reside import move_to_planet, MoveToPlanetInput


@pytest.mark.asyncio
async def test_graphql_out_planet_hello() -> None:
    from src.ui.graphql.main import app

    async with TestClient(app) as client:
        query = '{residence(residentName: "Joe"){welcomeMsg}}'
        result = await client.post(
            "/graphql", json={"query": query}
        ).json()
        assert result["data"]["residence"]["welcomeMsg"] == "Welcome to Tatooine!

yields:

self = <ariadne.asgi.GraphQL object at 0x7ffb8c0bdf40>, scope = {'asgi': {'version': '3.0'}, 'type': 'lifespan'}
receive = <bound method Queue.get of <Queue at 0x7ffb8c2b4c10 maxsize=0 _queue=[{'type': 'lifespan.startup'}] tasks=1>>
send = <bound method Queue.put of <Queue at 0x7ffb8c2b46d0 maxsize=0 tasks=1>>

Here is my app.py for reference

app,py

from typing import Optional
from ariadne import (
    make_executable_schema,
    MutationType,
    QueryType,
)
from ariadne.asgi import GraphQL
from apischema.graphql import graphql_schema
from graphql import print_schema

from src.core.reside import MoveToPlanet, move_to_planet, MoveToPlanetInput, residence

mutation = MutationType()

type_defs = print_schema(graphql_schema(query=[residence], mutation=[move_to_planet]))


@mutation.field("moveToPlanet")
def resolve_move_to_planet(
    *_: None, planetId: str, residentName: str
) -> Optional[MoveToPlanet]:
    return move_to_planet(
        MoveToPlanetInput(planet_id=planetId, resident_name=residentName)
    )


query = QueryType()


@query.field("residence")
def resolve_residence(*_: None, residentName: str) -> MoveToPlanet:
    return residence(residentName)


schema = make_executable_schema(
    type_defs, [query, mutation], snake_case_fallback_resolvers
)


app = GraphQL(schema)

Looks like this only supports asyncio at the moment. If you have any pointers I might be able to send a PR.

I'd like to test uploading files to API server.
Using your test client, I wrote the test as below:

with (datadir / "image.png").open("rb") as fp:
    files = {"image": ("sample.png", fp, "image/png")}
    response = await client.post("/api/upload_image", files=files)

But multipart file handling seems to have some error.
I don't think that decoding binary data of a file into str is possible.

        if isinstance(value, bytes):
>           value = value.decode("ascii")
E           UnicodeDecodeError: 'ascii' codec can't decode byte 0x89 in position 0: ordinal not in range(128)

../../../../../.venv/lib/python3.7/site-packages/async_asgi_testclient/multipart.py:59: UnicodeDecodeError

In the scenario where a websocket connection is closed before being accepted by the server, the WebSocketSession currently throws an AssertionError due to the received message not being type "websocket.accept" (see assertion)

If a connection request is cancelled before being accepted it should raise the message in someway so tests can process the response.

This should also apply to any message being received first that isn't an accept type since any of these would be a failure of the websocket protocol

Hi there and thanks for a great test client! It really saved me since Starlette's default TestClient is so awful.

But would it be possible to provide a nicer way to send headers with each request.
For example with requests you can do:

import requests
s = requests.Session()
s.headers.update({'my': 'header'})
s.get('/')

I couldn't find any similar functionality in your otherwise excellent testing client so I resorted to subclassing your TestClient like this:

from async_asgi_testclient import TestClient
class TestClient(TestClient):
	def __init__(self, *args, headers=None, **kwargs):
		super().__init__(*args, **kwargs)
		self.headers = headers

	async def open(
		self,
		path,
		*,
		method="GET",
		headers=None,
		data=None,
		form=None,
		query_string=None,
		json=None,
		scheme="http",
		cookies=None,
		stream=False,
		allow_redirects=True,
	):
		return await super().open(
			path,
			method=method,
			headers=self.headers,
			data=data,
			form=form,
			query_string=query_string,
			json=json,
			scheme=scheme,
			cookies=cookies,
			stream=stream,
			allow_redirects=allow_redirects
		)


client = TestClient(app)
client.headers = {
	'authorization': 'token my_token'
}

Is there maybe some better way of doing this that I am missing?
Or would you be willing to add a similar feature?
I can also open a pull request if you are open to that

Cookies from cookie jar are rendered in wrong way into Cookie request header. The HTTP request Cookie header should contain only cookie_name=cookie_value pairs delimited by semicolon (;). Currently whole cookie is rendered as for response Set-Cookie which cause than tested application sees cookies like Expires, Domain, Samesite etc...
Moreover if more than one cookie is in cookie jar, generated header makes whole request malformed as rendered cookies are \r\n separated (which obviously ends request's Cookie header at first occurrence and the rest are rubbish)

issue is here: async_asgi_testclient/testing.py:

        if cookie_jar and cookie_jar.output(header=""):
            headers.add("Cookie", cookie_jar.output(header=""))

I will make a PR with a fix soon.

TestClient always calls the GET routes, when routes are added using fastapi.APIRouter().add_api_router() no matter what method was requested.

Execute the following program

import asyncio
from async_asgi_testclient import TestClient as AsyncTestClient
from fastapi.testclient import TestClient

from fastapi import APIRouter, FastAPI


def get():
    print("called get", end=', ')

def post():
    print("called post", end=', ')

def put():
    print("called put", end=', ')

def delete():
    print("called delete", end=', ')

def make_router():
    router = APIRouter()
    router.add_api_route('/', get, methods=['GET'])
    router.add_api_route('/', post, methods=['POST'])
    router.add_api_route('/', put, methods=['PUT'])
    return router

api = FastAPI()
api.include_router(make_router(), prefix='/foo')


async def main():
    print("sync:")
    with TestClient(api) as client:
        print("GET request", client.get('/foo').status_code)
        print("POST request", client.post('/foo').status_code)
        print("PUT request", client.put('/foo').status_code)
        print("DELETE request", client.delete('/foo').status_code)
    print("async:")
    async with AsyncTestClient(api) as client:
        print("GET request", (await client.get('/foo')).status_code)
        print("POST request", (await client.post('/foo')).status_code)
        print("PUT request", (await client.put('/foo')).status_code)
        print("DELETE request", (await client.delete('/foo')).status_code)


asyncio.run(main())

output is

sync:
called get, GET request 200
called post, POST request 200
called put, PUT request 200
DELETE request 405
async:
called get, GET request 200
called get, POST request 200
called get, PUT request 200
called get, DELETE request 200

expected output is

sync:
called get, GET request 200
called post, POST request 200
called put, PUT request 200
DELETE request 405
async:
called get, GET request 200
called post, POST request 200
called put, PUT request 200
DELETE request 405

Python versions tested: 3.10, 3.11, 3.12
fastapi: 0.104.1
async-asgi-testclient: 1.4.11

when starlette returns a redirect response (such as when attempting to fix trailing slashes), it returns an absolute path, but async-asgi-testclient always interprets it as a relative path.
I propose that whenever the testclient recieves a redirect, it strip the hostname off the location, if any are found.

https://asgi.readthedocs.io/en/latest/specs/www.html#websocket-connection-scope

"subprotocols" key should be built from Sec-WebSocket-Protocol header value. Currently it's always empty.

        scope = {
            "type": "websocket",
            "headers": flatten_headers(headers),
            "path": path,
            "query_string": query_string_bytes,
            "root_path": "",
            "scheme": "http",
            "subprotocols": [],
        }

Hello,

Thanks for your library but it seems it has some problems which has to be resolved

WebSocket URL with query params does not work

#!/usr/bin/env python3
  
import asyncio
from fastapi import FastAPI
from starlette.websockets import WebSocket
from async_asgi_testclient import TestClient

app = FastAPI()

url1 = '/ws' # works ok
url2 = '/ws?token=token'  # failed


@app.websocket_route('/ws')
async def websocket_endpoint(websocket: WebSocket):

    await websocket.accept()
    await websocket.send_text('Hello')


async def main():
    async with TestClient(app) as client:
        async with client.websocket_connect(url2) as websocket:
            print(await websocket.receive_text())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Hi,

First of all, thanks for this. I was able to speed up our tests by an order of magnitude using this package.

I use it in somewhat unorthodox manner, by creating several TestClient instances for a few services I want to mock in tests, then patch aiohttp.ClientSession so that I can intercept outgoing calls made by my (non-web) application and route them to one of the TestClients.

Unfortunately, the signature of TestClient's http methods is a bit different than corresponding methods in aiohttp.ClientSession. To forward these calls, I need to frob incoming arguments before passing them to TestClient, and then wrap resulting Response object into async context managers and other shenanigans.

I guess TestClient's API was designed to match requests, not aiohttp, is that right?

If so, what do you think about adding a compatibility layer that would match aiohttp.ClientSession API?

First, thank you for creating this! This package is the only testing utility I've found that can consume and test an asgi streaming response (fastapi/starlette).

Inside this package, with newer versions of starlette, I do see some failures with some of the streaming tests.

Starting in 0.13.3

• test_upload_stream_from_download_stream

Then in 0.13.4, the same test starts to hang, instead of outright fail.

In 0.13.5, this test starts to hang as well.

• test_request_stream

In the latest version of starlette, 0.21.0, test_request_stream hangs, while test_upload_stream_from_download_stream fails with the following error:

$ pytest async_asgi_testclient/tests/test_testing.py::test_upload_stream_from_download_stream

============================= test session starts ==============================
platform darwin -- Python 3.10.6, pytest-7.1.3, pluggy-1.0.0
rootdir: /Users/bfalk/repos/async-asgi-testclient
plugins: anyio-3.6.1, asyncio-0.19.0, cov-4.0.0
asyncio: mode=strict
collected 1 item

async_asgi_testclient/tests/test_testing.py F                            [100%]

=================================== FAILURES ===================================
___________________ test_upload_stream_from_download_stream ____________________

starlette_app = <starlette.applications.Starlette object at 0x104bf5750>

    @pytest.mark.asyncio
    async def test_upload_stream_from_download_stream(starlette_app):
        from starlette.responses import StreamingResponse
    
        async def down_stream(request):
            def gen():
                for _ in range(3):
                    yield b"X" * 1024
    
            return StreamingResponse(gen())
    
        async def up_stream(request):
            async def gen():
                async for chunk in request.stream():
                    yield chunk
    
            return StreamingResponse(gen())
    
        starlette_app.add_route("/download_stream", down_stream, methods=["GET"])
        starlette_app.add_route("/upload_stream", up_stream, methods=["POST"])
    
        async with TestClient(starlette_app) as client:
            resp = await client.get("/download_stream", stream=True)
            assert resp.status_code == 200
            resp2 = await client.post(
                "/upload_stream", data=resp.iter_content(1024), stream=True
            )
            chunks = [c async for c in resp2.iter_content(1024)]
>           assert len(b"".join(chunks)) == 3 * 1024
E           AssertionError: assert 1024 == (3 * 1024)
E            +  where 1024 = len(b'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX...XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX')
E            +    where b'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX...XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' = <built-in method join of bytes object at 0x1030d4030>([b'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX...XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', b'', b''])
E            +      where <built-in method join of bytes object at 0x1030d4030> = b''.join

async_asgi_testclient/tests/test_testing.py:516: AssertionError
=========================== short test summary info ============================
FAILED async_asgi_testclient/tests/test_testing.py::test_upload_stream_from_download_stream
============================== 1 failed in 0.10s ===============================

I have this (admittedly broken) entry in my pyproject.toml:

[tool.poetry.scripts]
start = "poetry run uvicorn server:app --reload"

Adding and removing other deps works just fine, but when I tried to poetry add --dev async-asgi-testclient the addition crashed with:

Using version ^1.4.6 for async-asgi-testclient
...
  • Installing async-asgi-testclient (1.4.6): Failed
...
      Traceback (most recent call last):
        File "/Users/xxx-py3.9/lib/python3.8/site-packages/pkg_resources/__init__.py", line 2848, in get_entry_map
          ep_map = self._ep_map
        File "/Users/xxx-py3.9/lib/python3.8/site-packages/pkg_resources/__init__.py", line 2810, in __getattr__
          raise AttributeError(attr)
      AttributeError: _ep_map

      During handling of the above exception, another exception occurred:

      Traceback (most recent call last):
       File "/Users/xxx-py3.9/lib/python3.8/site-packages/pkg_resources/__init__.py", line 2495, in parse
          raise ValueError(msg, src)
      ValueError: ("EntryPoint must be in 'name=module:attrs [extras]' format", 'start=poetryrunuvicornserver:app--reload')

Normally I'd blame poetry, but somehow only this package triggers this, so 🤷🏿

P.S. package gets installed correctly into a clean/healthy project.
I'm still a little confused why installing this particular package trips on the "outside" of it...

According to asgi document, the websocket connection scheme should be ws or wss, but get http

This cause test failure for quart 0.15+/0.16+ (Error BadRequest)

Reference: https://asgi.readthedocs.io/en/latest/specs/www.html#websocket-connection-scope

First of all, thanks for developing such a useful library.

I use cookies for authentication, and accept WebSocket connection from the authenticated user.
I can set cookies to a TestClient instance as bellow, although, the cookies are not included in WebSocket connection request.

TestClient.websocket_connect() method has extra_headers argument, but adding cookies directly to HTTP Header is not easy.

So It would be helpful if WebSocketSession could take over cookies from TestClient.

async with TestClient(app) as client
    ck = SimpleCookie()
    ck["foo"] = "bar"
    client.cookie_jar = ck

    async with client.websocket_connect(endpoint) as ws:
        ...

Hi! Thanks for this project. I think it's a good way forward to have a framework-agnostic async test client for ASGI apps.

I see the client currently uses requests — would it make sense to switch to requests-async?

I don't think it'll provide any functional changes and maybe this will just be a maintenance cost. I'm just throwing the idea off my head. :-)

I'm testing an app that needs to know whether a WebSocket connection was done over ws:// or wss://. Currently, while I can override the scheme for HTTP requests, I cannot set the scheme used for WebSocket connections. Moreover, overriding this behavior via subclassing for example is quite difficult as the scheme is hard-coded inside the WebSocketSession.connect() method which is quite extensive.

It would be helpful if I could override the WS connection scheme and set it to wss:// or ws:// explicitly.