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
2 changes: 1 addition & 1 deletion .github/workflows/ci.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ jobs:
- uses: actions/setup-python@v6
with:
python-version: "3.14"
- run: pip install --group dev --group lint -e .
- run: pip install --group lint -e .
- run: pre-commit run --all-files
- run: mypy looptime --strict

Expand Down
4 changes: 2 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -180,12 +180,12 @@ async def test_me():
The markers can also be artificially injected by plugins/hooks if needed:

```python
import asyncio
import inspect
import pytest

@pytest.hookimpl(hookwrapper=True)
def pytest_pycollect_makeitem(collector, name, obj):
if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj):
if collector.funcnamefilter(name) and inspect.iscoroutinefunction(obj):
pytest.mark.asyncio(obj)
pytest.mark.looptime(end=60)(obj)
yield
Expand Down
61 changes: 10 additions & 51 deletions looptime/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
- A single event loop can be shared by multiple (but not all) tests.
- A single test can be spread over multiple (but not all) event loops.

An classic example:
A classic example:

- A session-scoped fixture ``server`` starts a port listener & an HTTP server.
- A module-scoped fixture ``data`` populates the server via POST requests.
Expand Down Expand Up @@ -52,7 +52,7 @@
Previously, the higher-scoped fixtures did not exist, so nothing breaks.

2. If the start time is explicitly defined and is in the future, move the time
forwards as specified — indistinguishable from the previous behaviour
forward as specified — indistinguishable from the previous behaviour
(except there could be artifacts from the previous tests in the loop).

3. If the start time is explicitly defined and is in the past, issue a warning
Expand Down Expand Up @@ -157,10 +157,6 @@ def pytest_addoption(parser: Any) -> None:
help="Run unmarked tests with the fake loop time by default.")


EventLoopScopes = dict[str, list[str]] # {fixture_name -> [outer_scopes, …, innermost_scope]}
EVENT_LOOP_SCOPES = pytest.StashKey[EventLoopScopes]()


@pytest.hookimpl(wrapper=True)
def pytest_fixture_setup(fixturedef: pytest.FixtureDef[Any], request: pytest.FixtureRequest) -> Any:
# Setup as usual. We do the magic only afterwards, when we have the event loop created.
Expand All @@ -182,19 +178,12 @@ def pytest_fixture_setup(fixturedef: pytest.FixtureDef[Any], request: pytest.Fix
else:
is_bp_runner = isinstance(result, bp_Runner)

# Patch the event loop at creation — even if unused and not enabled. We cannot patch later
# in the middle of the run: e.g. for a session-scoped loop used in a few tests out of many.
# NB: For the lowest "function" scope, we still cannot decide which options to use, since
# we do not know yet if it will be the running loop or not — so we cannot optimize here
# in order to patch-and-configure only once; we must patch here & configure+activate later.
if should_patch and (is_loop or is_runner or is_bp_runner):

# Populate the helper mapper of names-to-scopes, as used in the test hook below.
if EVENT_LOOP_SCOPES not in request.session.stash:
request.session.stash[EVENT_LOOP_SCOPES] = {}
event_loop_scopes: EventLoopScopes = request.session.stash[EVENT_LOOP_SCOPES]
event_loop_scopes.setdefault(fixturedef.argname, []).append(fixturedef.scope)

# Patch the event loop at creation — even if unused and not enabled. We cannot patch later
# in the middle of the run: e.g. for a session-scoped loop used in a few tests out of many.
# NB: For the lowest "function" scope, we still cannot decide which options to use, since
# we do not know yet if it will be the running loop or not — so we cannot optimize here
# in order to patch-and-configure only once; we must patch here & configure+activate later.
if isinstance(result, asyncio.BaseEventLoop):
patchers.patch_event_loop(result, _enabled=False)
elif sys.version_info >= (3, 11) and isinstance(result, asyncio.Runner):
Expand All @@ -212,39 +201,6 @@ def pytest_fixture_setup(fixturedef: pytest.FixtureDef[Any], request: pytest.Fix
return result


@pytest.hookimpl(wrapper=True)
def pytest_fixture_post_finalizer(fixturedef: pytest.FixtureDef[Any], request: pytest.FixtureRequest) -> Any:
# Cleanup the helper mapper of the fixture's names-to-scopes, as used in the test-running hook.
# Internal consistency check: some cases should not happen, but we do not fail if they do.
should_patch = _should_patch(fixturedef, request)
if should_patch and EVENT_LOOP_SCOPES in request.session.stash:
event_loop_scopes: EventLoopScopes = request.session.stash[EVENT_LOOP_SCOPES]
if fixturedef.argname not in event_loop_scopes:
warnings.warn(
f"Fixture {fixturedef.argname!r} not found in the cache of scopes."
f" Report as a bug, please add a reproducible snippet.",
RuntimeWarning,
)
elif not event_loop_scopes[fixturedef.argname]:
warnings.warn(
f"Fixture {fixturedef.argname!r} has the empty cache of scopes."
f" Report as a bug, please add a reproducible snippet.",
RuntimeWarning,
)
elif event_loop_scopes[fixturedef.argname][-1] != fixturedef.scope:
warnings.warn(
f"Fixture {fixturedef.argname!r} has the broken cache of scopes:"
f" {event_loop_scopes[fixturedef.argname]!r}, expecting {fixturedef.scope!r}"
f" Report as a bug, please add a reproducible snippet.",
RuntimeWarning,
)
else:
event_loop_scopes[fixturedef.argname][-1:] = []

# Go as usual.
return (yield)


# This hook is the latest (deepest) possible entrypoint before diving into the test function itself,
# with all the fixtures executed earlier, so that their setup time is not taken into account.
# Here, we know the actual running loop (out of many) chosen by pytest-asyncio & its marks/configs.
Expand Down Expand Up @@ -305,6 +261,9 @@ def _should_patch(fixturedef: pytest.FixtureDef[Any], request: pytest.FixtureReq
return True

# pytest-asyncio>=1.0.0 exposes several event loops, one per scope, all hidden in the module.
# We patch BOTH the default implementation, AND all those dirty hacks that users might make.
# NB: We also report True on unrelated fixtures, such as `unused_tcp_port_factory`, etc.
# This has no effect: they will not pass the extra test on being patchable loops & runners.
asyncio_plugin = request.config.pluginmanager.getplugin("asyncio") # a module object
asyncio_names: set[str] = {
name for name in dir(asyncio_plugin) if _is_fixture(getattr(asyncio_plugin, name))
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ dev = [
"pytest-mock",
]
lint = [
{include-group = "dev"},
"mypy==1.19.0",
"pre-commit",
"isort",
Expand Down
Loading