Can't run multiple tests with same instance of ReusableClient

[slavict]

How to reproduce

import pytest
from sanic_testing.reusable import ReusableClient
from sanic import Sanic
from sanic import response


@pytest.fixture(scope="session")
def app():
    sanic_app = Sanic("TestSanic")

    @sanic_app.get("/")
    def basic(request):
        return response.text("foo")

    @sanic_app.post("/api/login")
    def basic(request):
        return response.text("foo")

    @sanic_app.get("/api/resources")
    def basic(request):
        return response.text("foo")

    return sanic_app


@pytest.fixture(scope="session")
def cli(app):
    cli = ReusableClient(app)
    return cli


def test_root(cli):
    with cli:
        _, response = cli.get("/")
        assert response.status == 200


def test_login(cli):
    with cli:
        _, response = cli.post("/api/login", )
        assert response.status == 200

        _, response = cli.get("/api/resources")
        assert response.status == 200

Error that I got

self = <_UnixSelectorEventLoop running=False closed=False debug=False>
future = <Task finished name='Task-18' coro=<StartupMixin.create_server() done, defined at /home/ghost/wuw2/lib/python3.10/site-packages/sanic/mixins/startup.py:347> exception=RuntimeError('cannot reuse already awaited coroutine')>

    def run_until_complete(self, future):
        """Run until the Future is done.
    
        If the argument is a coroutine, it is wrapped in a Task.
    
        WARNING: It would be disastrous to call run_until_complete()
        with the same coroutine twice -- it would wrap it in two
        different Tasks and that can't be good.
    
        Return the Future's result, or raise its exception.
        """
        self._check_closed()
        self._check_running()
    
        new_task = not futures.isfuture(future)
        future = tasks.ensure_future(future, loop=self)
        if new_task:
            # An exception is raised if the future didn't complete, so there
            # is no need to log the "destroy pending task" message
            future._log_destroy_pending = False
    
        future.add_done_callback(_run_until_complete_cb)
        try:
            self.run_forever()
        except:
            if new_task and future.done() and not future.cancelled():
                # The coroutine raised a BaseException. Consume the exception
                # to not log a warning, the caller doesn't have access to the
                # local task.
                future.exception()
            raise
        finally:
            future.remove_done_callback(_run_until_complete_cb)
        if not future.done():
            raise RuntimeError('Event loop stopped before Future completed.')
    
>       return future.result()
E       RuntimeError: cannot reuse already awaited coroutine

[slavict]

Maybe that I did't understand how to use 'ReusableClient' class in right way, If that , I will be very grateful to see more complex example in documentation or at least in tests.

[ahopkins]

Maybe that I did't understand how to use 'ReusableClient' class in right way, If that , I will be very grateful to see more complex example in documentation or at least in tests.

I'll post an example here and then to the docs later today.

[pyx]

Any update on this? I am writing tests for setting cookies, the default app.test_client doesn't work.

I try to monkey-patch the application, replacing app.test_client with ReusableTestClient, and got AttributeError: Setting variables on Sanic instances is not allowed. You s.... I know how to circumvent that as I have 2 decades of python experience (with dunder methods trick), but I don't think it's the correct way...

It worked before (sanic<=20.x.x, IIRC), basically it allows me to pass in cookies=cookies to simulate user session. see: https://github.com/pyx/sanic-cookiesession/blob/9ea4491e1ba63496d8fd6dd9e18deb9aa22a8fb0/tests/test_session.py#L38

[kserhii]

I was able to create a ReusableTestClient fixture. My environment is sanic==23.12.1 and python3.12.

Here is how I I did this

# conftest.py

from my_app import create_app


@pytest.fixture
def config():
    return {'DEBUG': True}


@pytest.fixture
def app(config):
    Sanic.test_mode = True
    return create_app(config)


@pytest.fixture
def test_cli(app, event_loop):
    with ReusableClient(app, loop=event_loop) as cli:
        try:
            yield cli
        finally:
            event_loop.run_until_complete(cli._session.aclose())  # close request

where event_loop is the default loop from pytest-asyncio.

There is one issue with the ReusableClient. It doesn't close the connection correctly. First it closes the server socket and only then closes the HTTP client session. This creates a deadlock as server waits for all client connections to be closed. As you can see I added a workaround solution for this.

Here is a sample how this fixture can be used

# test_auth.py

def test_auth_token_missing(test_cli):
    req, resp = test_cli.get('/api/v1/sample.png')

    assert resp.status == 401
    assert resp.json == {
        'details': 'Authentication credentials were not provided'
    }

You can also create test client with predefined headers

@pytest.fixture
def auth_cli(app, token, event_loop):
    with ReusableClient(
            app, 
            loop=event_loop, 
            client_kwargs={'headers': {'Authorization': f'Bearer {token}'}}
    ) as cli:
        try:
            yield cli
        finally:
            event_loop.run_until_complete(cli._session.aclose())  # close request

If you need to run something async you can use event_loop.run_until_complete wrapper for this

def test_reader_cache(app, event_loop, auth_cli):
    req, resp = auth_cli.get('/api/v1/sample.png')

    assert resp.status == 200
    assert len(app.ctx.reader.cache) == 1

    event_loop.run_until_complete(asyncio.sleep(0.4))  # make sure cleanup task has been called

    assert len(app.ctx.reader.cache) == 0