Skip to content
Merged
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
4 changes: 2 additions & 2 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ jobs:
- name: Run tests
run: |
uv run pytest -vv tests --cov=anydi
uv run pytest -vv tests/ext/test_pytest_plugin.py --cov=anydi --cov-append --cov-report=xml -p anydi
uv run pytest -vv tests/ext/test_pytest_plugin.py --cov=anydi --cov-append --cov-report=xml -p anydi -p pytester
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
if: matrix.python == '3.10'
Expand Down Expand Up @@ -105,7 +105,7 @@ jobs:
- name: Run tests
run: |
uv run --no-sync pytest tests
uv run --no-sync pytest tests/ext/test_pytest_plugin.py -p anydi
uv run --no-sync pytest tests/ext/test_pytest_plugin.py -p anydi -p pytester

benchmarks:
name: Run benchmarks
Expand Down
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -15,4 +15,4 @@ fmt: ## Run code formatters

test: ## Run unit tests
uv run pytest -vv tests --cov=anydi
uv run pytest -vv tests/ext/test_pytest_plugin.py --cov=anydi --cov-append -p anydi
uv run pytest -vv tests/ext/test_pytest_plugin.py --cov=anydi --cov-append -p anydi -p pytester
173 changes: 137 additions & 36 deletions anydi/ext/pytest_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,23 @@
logger = logging.getLogger(__name__)


CONTAINER = pytest.StashKey[Container]()
# Containers this plugin put into test mode, switched back off at session end
_ACTIVATED = pytest.StashKey[list[Container]]()
# The global container the plugin replaced, restored at session end
_REPLACED_GLOBAL = pytest.StashKey[Container | None]()


def pytest_addoption(parser: pytest.Parser) -> None:
parser.addini(
"anydi",
help=(
"Wire the tests through the configured container: test mode, "
"the `container` fixture, injection"
),
type="bool",
default=True,
)
parser.addini(
"anydi_container",
help=(
Expand All @@ -42,31 +58,101 @@ def pytest_addoption(parser: pytest.Parser) -> None:
)


@pytest.hookimpl(hookwrapper=True)
def pytest_fixture_setup(
fixturedef: pytest.FixtureDef[Any], request: SubRequest
) -> Generator[None]:
"""Automatically enable test mode on the container fixture."""
yield
if fixturedef.argname == "container" and fixturedef.cached_result is not None:
@pytest.hookimpl(trylast=True)
def pytest_configure(config: pytest.Config) -> None:
if not config.getini("anydi"):
return
config.stash[_ACTIVATED] = []
container = _import_configured_container(config)
if container is not None:
_activate(config, container)
config.pluginmanager.register(_AnydiPlugin(), "anydi-fixtures")


def pytest_unconfigure(config: pytest.Config) -> None:
for container in config.stash.get(_ACTIVATED, []):
container.disable_test_mode()
if _REPLACED_GLOBAL in config.stash:
reset_global_container()
previous = config.stash[_REPLACED_GLOBAL]
if previous is not None:
set_global_container(previous)


class _AnydiPlugin:
"""Hooks and fixtures registered only when `anydi` is enabled."""

@pytest.hookimpl(trylast=True)
def pytest_collection_finish(self, session: pytest.Session) -> None:
# Collection has imported the application, so a global container
# created there exists now, before any fixture can override it
config = session.config
container = config.stash.get(CONTAINER, None) or get_global_container_or_none()
if container is not None:
_activate(config, container)
_use_as_global(config, container)

@pytest.hookimpl(hookwrapper=True)
def pytest_fixture_setup(
self, fixturedef: pytest.FixtureDef[Any], request: SubRequest
) -> Generator[None]:
"""Enable test mode on a container provided by a `container` fixture."""
yield
if fixturedef.argname != "container" or fixturedef.cached_result is None:
return
container = fixturedef.cached_result[0]
if isinstance(container, Container):
container.enable_test_mode()
# Let global references resolve against the container under test,
# without introducing a global container the application never used
if uses_global_container():
reset_global_container()
set_global_container(container)


@pytest.fixture(scope="session")
def container(request: pytest.FixtureRequest) -> Container:
"""Container fixture."""
return _find_container(request)
if not isinstance(container, Container):
return
if container is request.config.stash.get(CONTAINER, None):
return
container.enable_test_mode()
# Let global references resolve against the container under test,
# without introducing a global container the application never used
if uses_global_container():
reset_global_container()
set_global_container(container)

@pytest.fixture(scope="session")
def container(self, request: pytest.FixtureRequest) -> Container:
"""Container fixture."""
container = request.config.stash.get(CONTAINER, None)
if container is not None:
return container
return _find_container(request)

@pytest.fixture(autouse=True)
def _anydi_inject(self, request: pytest.FixtureRequest) -> None:
"""Inject dependencies into sync test functions."""
_inject(request)

@pytest.fixture(autouse=True)
def _anydi_ainject(self, request: pytest.FixtureRequest) -> None:
"""Inject dependencies into async test functions."""
_ainject(request)


def _activate(config: pytest.Config, container: Container) -> None:
"""Put the container into test mode for the rest of the session."""
config.stash[CONTAINER] = container
# A container already in test mode belongs to whoever enabled it
if not container._test_mode:
container.enable_test_mode()
config.stash[_ACTIVATED].append(container)


def _use_as_global(config: pytest.Config, container: Container) -> None:
"""Point global references at the container under test."""
if not uses_global_container():
return
previous = get_global_container_or_none()
if previous is container:
return
reset_global_container()
set_global_container(container)
config.stash.setdefault(_REPLACED_GLOBAL, previous)


@pytest.fixture(autouse=True)
def _anydi_inject(request: pytest.FixtureRequest) -> None:
def _inject(request: pytest.FixtureRequest) -> None:
"""Inject dependencies into sync test functions."""
if inspect.iscoroutinefunction(request.function):
return
Expand All @@ -86,8 +172,7 @@ def _anydi_inject(request: pytest.FixtureRequest) -> None:
logger.warning("Failed to resolve '%s' for %s", name, request.node.nodeid)


@pytest.fixture(autouse=True)
def _anydi_ainject(request: pytest.FixtureRequest) -> None:
def _ainject(request: pytest.FixtureRequest) -> None:
"""Inject dependencies into async test functions."""
if not inspect.iscoroutinefunction(
request.function
Expand Down Expand Up @@ -190,25 +275,41 @@ def _extract_dependency_type(annotation: Any) -> tuple[Any, bool]:
return annotation, False


def _find_container(request: pytest.FixtureRequest) -> Container:
"""Find container from config or auto-detection."""
container_path = cast(str | None, request.config.getini("anydi_container"))
if container_path:
try:
return import_container(container_path)
except ImportError as exc:
raise RuntimeError(
f"Failed to load container from config "
f"'anydi_container={container_path}': {exc}"
) from exc
def _import_configured_container(config: pytest.Config) -> Container | None:
"""Import the container named by `anydi_container`, if set."""
container_path = cast(str | None, config.getini("anydi_container"))
if not container_path:
return None
try:
return import_container(container_path)
except ImportError as exc:
raise pytest.UsageError(
f"Failed to load container from config "
f"'anydi_container={container_path}': {exc}"
) from exc


def _detect_container(config: pytest.Config) -> Container | None:
"""Find the container from config, the global one or `anydi_django`."""
container = _import_configured_container(config)
if container is not None:
return container

global_container = get_global_container_or_none()
if global_container is not None:
return global_container

pluginmanager = request.config.pluginmanager
pluginmanager = config.pluginmanager
if pluginmanager.hasplugin("django") and importlib.util.find_spec("anydi_django"):
return import_container("anydi_django.container")
return None


def _find_container(request: pytest.FixtureRequest) -> Container:
"""Find container from config or auto-detection."""
container = _detect_container(request.config)
if container is not None:
return container

raise pytest.FixtureLookupError(
None,
Expand Down
34 changes: 34 additions & 0 deletions docs/usage/testing.md
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,40 @@ def container() -> Container:

**Note:** The fixture takes priority over the configuration if both are defined.

#### Test mode

The plugin puts the container in test mode before the first fixture runs, so a fixture can override through the container your application imports, without requesting the `container` fixture:

```python
# conftest.py
from collections.abc import Iterator
from unittest import mock

import pytest

from myapp.container import container
from myapp.payments import PaymentClient


@pytest.fixture(scope="session", autouse=True)
def payment_client_mock() -> Iterator[mock.MagicMock]:
payment_client_mock = mock.MagicMock(spec=PaymentClient)
with container.override(PaymentClient, payment_client_mock):
yield payment_client_mock
```

This works for the container named by `anydi_container` and for a global container that exists once the tests are collected. The plugin turns test mode off again when the session ends. A container from your own `container` fixture enters test mode when the fixture first runs.

#### Turning the plugin off

The plugin is on wherever `AnyDI` is installed. To turn it off in a project, set `anydi = false`. The plugin then adds no `container` fixture, injects nothing and leaves the container untouched:

```toml
# pyproject.toml
[tool.pytest.ini_options]
anydi = false
```

### Usage

#### Explicit injection with `Provide[T]`
Expand Down
Loading
Loading