Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/how-to-guides/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -20,5 +20,6 @@ How-To Guides
parametrize_with_asyncio
uvloop
test_item_is_async
isolate_async_apps

This section of the documentation provides code snippets and recipes to accomplish specific tasks with pytest-asyncio.
60 changes: 60 additions & 0 deletions docs/how-to-guides/isolate_async_apps.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
===============================================
How to isolate async applications in your tests
===============================================

Async web frameworks (FastAPI, Quart, aiohttp) are typically tested by importing an ``app`` object and overriding its dependencies. Two subtle isolation traps can cause tests to pass in isolation but fail in a suite, or leak state across tests.

This guide shows how to use ``pytest-asyncio`` fixtures to avoid both.


Trap 1: Stale module-level imports after reload
-----------------------------------------------

If your test module imports the application at the top level:

.. code-block:: python

# test_a.py
from api.main import app # module-level import

@pytest.mark.asyncio
async def test_a():
...

and another test reloads the module to pick up configuration changes:

.. code-block:: python

# test_b.py
import importlib
import api.main
importlib.reload(api.main) # creates a NEW app object

then ``test_a`` still holds a reference to the *old* ``app`` object, while ``test_b`` sees the *new* one. Dependency overrides and state diverge silently.

**Fix:** Import ``app`` inside the fixture or test function so you always get the current module state:

.. include:: isolate_async_apps_example.py
:code: python
:start-after: # -- FIX 1: fixture-scoped import
:end-before: # -- FIX 2: temporary directories

Trap 2: Default file-system paths in constructors
-------------------------------------------------

An async service that accepts an optional path:

.. code-block:: python

class UploadService:
def __init__(self, upload_dir: Path = UPLOAD_DIR):
self.upload_dir = upload_dir

will reuse the same default directory across every test that forgets to override it. Files written by one test are visible to the next.

**Fix:** Always inject a temporary directory via a fixture, even when the parameter is "optional":

.. include:: isolate_async_apps_example.py
:code: python
:start-after: # -- FIX 2: temporary directories
:end-before: # -- end of example
133 changes: 133 additions & 0 deletions docs/how-to-guides/isolate_async_apps_example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
"""Example patterns for isolating async applications in tests.

Run with: pytest isolate_async_apps_example.py -v
"""

from __future__ import annotations

import tempfile
from pathlib import Path

import pytest


# ---------------------------------------------------------------------------
# Demo application (pretend this is api/main.py in your project)
# ---------------------------------------------------------------------------

class FakeUploadService:
"""Async service with a mutable default path."""

def __init__(self, upload_dir: Path | None = None) -> None:
self.upload_dir = upload_dir or Path("/tmp/uploads")
self.upload_dir.mkdir(parents=True, exist_ok=True)

async def save(self, name: str, data: bytes) -> Path:
dest = self.upload_dir / name
dest.write_bytes(data)
return dest


class FakeApp:
"""Minimal async app that owns a service."""

def __init__(self, upload_dir: Path | None = None) -> None:
self.upload_service = FakeUploadService(upload_dir=upload_dir)


# Global singleton-like app (common in FastAPI patterns)
app = FakeApp()


# -- FIX 1: fixture-scoped import


@pytest.fixture
async def fresh_app():
"""Import app inside the fixture so reloads are visible.

If another test calls ``importlib.reload(api.main)``, this fixture
will see the new ``app`` object on its next invocation.
"""
# In a real project: from api.main import app
import __main__ as _api_main # stand-in for the real module

# If you need to mutate the app (dependency overrides, etc.),
# do it here and yield, then clean up in teardown.
_api_main.app.upload_service = FakeUploadService()
yield _api_main.app
# Teardown: reset any overrides so the next test starts clean
_api_main.app.upload_service = FakeUploadService()


@pytest.mark.asyncio
async def test_upload_with_fresh_app(fresh_app: FakeApp):
"""Uses the fixture-scoped app import."""
path = await fresh_app.upload_service.save("a.txt", b"hello")
assert path.read_bytes() == b"hello"


@pytest.mark.asyncio
async def test_app_is_reloaded(fresh_app: FakeApp):
"""After a reload, the fixture still sees the current app."""
import importlib

import __main__ as _api_main

# Simulate another test mutating the module
old_app = _api_main.app
_api_main.app = FakeApp() # new instance
importlib.reload(_api_main) # or reload(api.main) in real code

# The fixture should now serve the new app
current = fresh_app
assert current is not old_app


# -- FIX 2: temporary directories


@pytest.fixture
async def upload_service(tmp_path: Path):
"""Service backed by a unique temp dir per test."""
service = FakeUploadService(upload_dir=tmp_path / "uploads")
yield service
# tmp_path is automatically cleaned up by pytest


@pytest.mark.asyncio
async def test_upload_isolated(upload_service: FakeUploadService):
path = await upload_service.save("b.txt", b"world")
assert path.read_bytes() == b"world"


@pytest.mark.asyncio
async def test_upload_does_not_bleed(upload_service: FakeUploadService):
"""The previous test's file should not exist here."""
files = list(upload_service.upload_dir.iterdir())
assert files == [] # passes because each test gets its own tmp_path


# ---------------------------------------------------------------------------
# Combined pattern: async app + temp dir in one fixture
# ---------------------------------------------------------------------------


@pytest.fixture
async def isolated_app(tmp_path: Path):
"""Full isolation: fresh import + fresh filesystem."""
import __main__ as _api_main

app = _api_main.app
app.upload_service = FakeUploadService(upload_dir=tmp_path / "uploads")
yield app
app.upload_service = FakeUploadService()


@pytest.mark.asyncio
async def test_end_to_end(isolated_app: FakeApp):
path = await isolated_app.upload_service.save("c.txt", b"combined")
assert path.read_bytes() == b"combined"


# -- end of example
Loading