In WebSocketSession.connect() method, check if self.cookies is None to use TestClient.cookie_jar. But self.cookies is never be None because the line self.cookies = cookies or {} exists in its constructor.

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

FastAPI provides the key client in the scope for accessing ip,port tuple of the request origin. I lacked it for my tests, so added and think it might be usable for someone else.

P.S: Great repo, exactly what I lacked :)

This fixes issues with asyncio and garbage collection. Specifically, if a running task within asyncio doesn't have a hard reference, it can be garbage collected causing a number of strange errors like Task was destroyed but it is pending!. Holding a hard reference to running tasks solves this issue which is what this PR does.

To consistently replicate this error we can force garbage collection in between every step of the event loop as follows:

• Edit /usr/lib/python3.*/asyncio/base_events.py and within def _run_once, near the bottom immediately within the for i in range(ntodo):, add import gc; gc.collect()
• Make sure your testing framework is using asyncio (rather than uvloop for example) so that it runs the code we just modified (pytest does by default)

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 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:
        ...

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())

This is my fix for #47. It allows passing scheme= to a websocket connection. I also took the liberty of passing any arguments to websocket_connect down to the WebSocketSession constructor, similar to how all the other get, post etc. method just pass arguments to open.

Happy to fix if there is any feedback.

I made a fix which renders cookies correctly in request header. In the previous code cookie_jar.output(headers="") was rendering cookies as for Set-Cookie response header, which means that additional flags like Expires, Max-Age, SameSite etc was added.

Why it was working?

Because both tested framework - Quart and Starlette, in versions specified in requirements, uses same library and SimpleCookie to parse this header. So even with Set-Cookie syntax in the header, all tests were passing as SimpleCookie.load method supports both: Cookie and Set-Cookie

However, Starlette in the current version has custom cookie parsing mechanism and updating starlette cause tests failing.

What has been changed

I've changed Cookie request header rendering, so only cookie name and value are rendered now. Also multiple cookies are correctly separated by ; not \r\n. I've add a new /cookie-raw endpoint in both frameworks to make sure that Cookie header seen by the asgi app is correct. I was NOT updating Starlette.

This PR closes https://github.com/vinissimus/async-asgi-testclient/issues/41

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

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...

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 ===============================

Fixes #54

• Test classes can now receive the early termination event as a raised Exception where the body of the exception is the closure message (incase validation is wanted to be done on the response)

• Also added tests for quartz WebSocket endpoints as they were missing

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

WIP: test suite for ASGI specification compliance.

It seems like a great idea for a ASGI test library to have tests itself to check it is compliant with the spec, and if there are new versions that come out, will make it easier to update to match the spec.

There are a number of tests that currently xfail in this suite - one or two of those tests might be things that I'm wrong in my interpretation of the spec, but I'm mostly pretty confident in them.

I've intentionally made these xfails rather than just fixing the code in the same PR, so the idea would be to fix the issues one by one and remove the xfail, or in cases where the issue is a WONTFIX, change the test and/or indicate as such in the xfail reason.

There is still a need for a couple more tests in the websocket section, and probably could do with some more type-oriented negative tests (e.g. testing what happens if you send a message with the wrong type in the event values, like with 'status').

Of particular note, the scope['path'] test (test_http:test_http_path_is_not_escaped) I'm not entirely sure that test should exist, it might not be valid. More explanation in the comments. This one is one of the main things that makes me want to keep this PR draft a bit longer, both for comment and while I decide what to do with that test.

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.