+++ /dev/null
-Apache License
- Version 2.0, January 2004
- http://www.apache.org/licenses/
-
- TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
- 1. Definitions.
-
- "License" shall mean the terms and conditions for use, reproduction,
- and distribution as defined by Sections 1 through 9 of this document.
-
- "Licensor" shall mean the copyright owner or entity authorized by
- the copyright owner that is granting the License.
-
- "Legal Entity" shall mean the union of the acting entity and all
- other entities that control, are controlled by, or are under common
- control with that entity. For the purposes of this definition,
- "control" means (i) the power, direct or indirect, to cause the
- direction or management of such entity, whether by contract or
- otherwise, or (ii) ownership of fifty percent (50%) or more of the
- outstanding shares, or (iii) beneficial ownership of such entity.
-
- "You" (or "Your") shall mean an individual or Legal Entity
- exercising permissions granted by this License.
-
- "Source" form shall mean the preferred form for making modifications,
- including but not limited to software source code, documentation
- source, and configuration files.
-
- "Object" form shall mean any form resulting from mechanical
- transformation or translation of a Source form, including but
- not limited to compiled object code, generated documentation,
- and conversions to other media types.
-
- "Work" shall mean the work of authorship, whether in Source or
- Object form, made available under the License, as indicated by a
- copyright notice that is included in or attached to the work
- (an example is provided in the Appendix below).
-
- "Derivative Works" shall mean any work, whether in Source or Object
- form, that is based on (or derived from) the Work and for which the
- editorial revisions, annotations, elaborations, or other modifications
- represent, as a whole, an original work of authorship. For the purposes
- of this License, Derivative Works shall not include works that remain
- separable from, or merely link (or bind by name) to the interfaces of,
- the Work and Derivative Works thereof.
-
- "Contribution" shall mean any work of authorship, including
- the original version of the Work and any modifications or additions
- to that Work or Derivative Works thereof, that is intentionally
- submitted to Licensor for inclusion in the Work by the copyright owner
- or by an individual or Legal Entity authorized to submit on behalf of
- the copyright owner. For the purposes of this definition, "submitted"
- means any form of electronic, verbal, or written communication sent
- to the Licensor or its representatives, including but not limited to
- communication on electronic mailing lists, source code control systems,
- and issue tracking systems that are managed by, or on behalf of, the
- Licensor for the purpose of discussing and improving the Work, but
- excluding communication that is conspicuously marked or otherwise
- designated in writing by the copyright owner as "Not a Contribution."
-
- "Contributor" shall mean Licensor and any individual or Legal Entity
- on behalf of whom a Contribution has been received by Licensor and
- subsequently incorporated within the Work.
-
- 2. Grant of Copyright License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- copyright license to reproduce, prepare Derivative Works of,
- publicly display, publicly perform, sublicense, and distribute the
- Work and such Derivative Works in Source or Object form.
-
- 3. Grant of Patent License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- (except as stated in this section) patent license to make, have made,
- use, offer to sell, sell, import, and otherwise transfer the Work,
- where such license applies only to those patent claims licensable
- by such Contributor that are necessarily infringed by their
- Contribution(s) alone or by combination of their Contribution(s)
- with the Work to which such Contribution(s) was submitted. If You
- institute patent litigation against any entity (including a
- cross-claim or counterclaim in a lawsuit) alleging that the Work
- or a Contribution incorporated within the Work constitutes direct
- or contributory patent infringement, then any patent licenses
- granted to You under this License for that Work shall terminate
- as of the date such litigation is filed.
-
- 4. Redistribution. You may reproduce and distribute copies of the
- Work or Derivative Works thereof in any medium, with or without
- modifications, and in Source or Object form, provided that You
- meet the following conditions:
-
- (a) You must give any other recipients of the Work or
- Derivative Works a copy of this License; and
-
- (b) You must cause any modified files to carry prominent notices
- stating that You changed the files; and
-
- (c) You must retain, in the Source form of any Derivative Works
- that You distribute, all copyright, patent, trademark, and
- attribution notices from the Source form of the Work,
- excluding those notices that do not pertain to any part of
- the Derivative Works; and
-
- (d) If the Work includes a "NOTICE" text file as part of its
- distribution, then any Derivative Works that You distribute must
- include a readable copy of the attribution notices contained
- within such NOTICE file, excluding those notices that do not
- pertain to any part of the Derivative Works, in at least one
- of the following places: within a NOTICE text file distributed
- as part of the Derivative Works; within the Source form or
- documentation, if provided along with the Derivative Works; or,
- within a display generated by the Derivative Works, if and
- wherever such third-party notices normally appear. The contents
- of the NOTICE file are for informational purposes only and
- do not modify the License. You may add Your own attribution
- notices within Derivative Works that You distribute, alongside
- or as an addendum to the NOTICE text from the Work, provided
- that such additional attribution notices cannot be construed
- as modifying the License.
-
- You may add Your own copyright statement to Your modifications and
- may provide additional or different license terms and conditions
- for use, reproduction, or distribution of Your modifications, or
- for any such Derivative Works as a whole, provided Your use,
- reproduction, and distribution of the Work otherwise complies with
- the conditions stated in this License.
-
- 5. Submission of Contributions. Unless You explicitly state otherwise,
- any Contribution intentionally submitted for inclusion in the Work
- by You to the Licensor shall be under the terms and conditions of
- this License, without any additional terms or conditions.
- Notwithstanding the above, nothing herein shall supersede or modify
- the terms of any separate license agreement you may have executed
- with Licensor regarding such Contributions.
-
- 6. Trademarks. This License does not grant permission to use the trade
- names, trademarks, service marks, or product names of the Licensor,
- except as required for reasonable and customary use in describing the
- origin of the Work and reproducing the content of the NOTICE file.
-
- 7. Disclaimer of Warranty. Unless required by applicable law or
- agreed to in writing, Licensor provides the Work (and each
- Contributor provides its Contributions) on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- implied, including, without limitation, any warranties or conditions
- of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
- PARTICULAR PURPOSE. You are solely responsible for determining the
- appropriateness of using or redistributing the Work and assume any
- risks associated with Your exercise of permissions under this License.
-
- 8. Limitation of Liability. In no event and under no legal theory,
- whether in tort (including negligence), contract, or otherwise,
- unless required by applicable law (such as deliberate and grossly
- negligent acts) or agreed to in writing, shall any Contributor be
- liable to You for damages, including any direct, indirect, special,
- incidental, or consequential damages of any character arising as a
- result of this License or out of the use or inability to use the
- Work (including but not limited to damages for loss of goodwill,
- work stoppage, computer failure or malfunction, or any and all
- other commercial damages or losses), even if such Contributor
- has been advised of the possibility of such damages.
-
- 9. Accepting Warranty or Additional Liability. While redistributing
- the Work or Derivative Works thereof, You may choose to offer,
- and charge a fee for, acceptance of support, warranty, indemnity,
- or other liability obligations and/or rights consistent with this
- License. However, in accepting such obligations, You may act only
- on Your own behalf and on Your sole responsibility, not on behalf
- of any other Contributor, and only if You agree to indemnify,
- defend, and hold each Contributor harmless for any liability
- incurred by, or claims asserted against, such Contributor by reason
- of your accepting any such warranty or additional liability.
-
- END OF TERMS AND CONDITIONS
-
- APPENDIX: How to apply the Apache License to your work.
-
- To apply the Apache License to your work, attach the following
- boilerplate notice, with the fields enclosed by brackets "{}"
- replaced with your own identifying information. (Don't include
- the brackets!) The text should be enclosed in the appropriate
- comment syntax for the file format. We also recommend that a
- file or class name and description of purpose be included on the
- same "printed page" as the copyright notice for easier
- identification within third-party archives.
-
- Copyright {yyyy} {name of copyright owner}
-
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
+++ /dev/null
-"""pytest-asyncio implementation."""
-
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import contextvars
-import enum
-import functools
-import inspect
-import socket
-import sys
-import warnings
-from asyncio import AbstractEventLoop, AbstractEventLoopPolicy
-from collections.abc import (
- AsyncIterator,
- Awaitable,
- Coroutine as AbstractCoroutine,
- Generator,
- Iterable,
- Iterator,
- Sequence,
-)
-from typing import (
- Any,
- Callable,
- Literal,
- TypeVar,
- Union,
- cast,
- overload,
-)
-
-import pluggy
-import pytest
-from _pytest.scope import Scope
-from pytest import (
- Collector,
- Config,
- FixtureDef,
- FixtureRequest,
- Function,
- Item,
- Mark,
- Metafunc,
- Parser,
- PytestCollectionWarning,
- PytestDeprecationWarning,
- PytestPluginManager,
-)
-
-if sys.version_info >= (3, 10):
- from typing import ParamSpec
-else:
- from typing_extensions import ParamSpec
-
-
-_ScopeName = Literal["session", "package", "module", "class", "function"]
-_T = TypeVar("_T")
-_R = TypeVar("_R", bound=Union[Awaitable[Any], AsyncIterator[Any]])
-_P = ParamSpec("_P")
-FixtureFunction = Callable[_P, _R]
-
-
-class PytestAsyncioError(Exception):
- """Base class for exceptions raised by pytest-asyncio"""
-
-
-class Mode(str, enum.Enum):
- AUTO = "auto"
- STRICT = "strict"
-
-
-ASYNCIO_MODE_HELP = """\
-'auto' - for automatically handling all async functions by the plugin
-'strict' - for autoprocessing disabling (useful if different async frameworks \
-should be tested together, e.g. \
-both pytest-asyncio and pytest-trio are used in the same project)
-"""
-
-
-def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None:
- group = parser.getgroup("asyncio")
- group.addoption(
- "--asyncio-mode",
- dest="asyncio_mode",
- default=None,
- metavar="MODE",
- help=ASYNCIO_MODE_HELP,
- )
- parser.addini(
- "asyncio_mode",
- help="default value for --asyncio-mode",
- default="strict",
- )
- parser.addini(
- "asyncio_default_fixture_loop_scope",
- type="string",
- help="default scope of the asyncio event loop used to execute async fixtures",
- default=None,
- )
- parser.addini(
- "asyncio_default_test_loop_scope",
- type="string",
- help="default scope of the asyncio event loop used to execute tests",
- default="function",
- )
-
-
-@overload
-def fixture(
- fixture_function: FixtureFunction[_P, _R],
- *,
- scope: _ScopeName | Callable[[str, Config], _ScopeName] = ...,
- loop_scope: _ScopeName | None = ...,
- params: Iterable[object] | None = ...,
- autouse: bool = ...,
- ids: (
- Iterable[str | float | int | bool | None]
- | Callable[[Any], object | None]
- | None
- ) = ...,
- name: str | None = ...,
-) -> FixtureFunction[_P, _R]: ...
-
-
-@overload
-def fixture(
- fixture_function: None = ...,
- *,
- scope: _ScopeName | Callable[[str, Config], _ScopeName] = ...,
- loop_scope: _ScopeName | None = ...,
- params: Iterable[object] | None = ...,
- autouse: bool = ...,
- ids: (
- Iterable[str | float | int | bool | None]
- | Callable[[Any], object | None]
- | None
- ) = ...,
- name: str | None = None,
-) -> Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]]: ...
-
-
-def fixture(
- fixture_function: FixtureFunction[_P, _R] | None = None,
- loop_scope: _ScopeName | None = None,
- **kwargs: Any,
-) -> (
- FixtureFunction[_P, _R]
- | Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]]
-):
- if fixture_function is not None:
- _make_asyncio_fixture_function(fixture_function, loop_scope)
- return pytest.fixture(fixture_function, **kwargs)
-
- else:
-
- @functools.wraps(fixture)
- def inner(fixture_function: FixtureFunction[_P, _R]) -> FixtureFunction[_P, _R]:
- return fixture(fixture_function, loop_scope=loop_scope, **kwargs)
-
- return inner
-
-
-def _is_asyncio_fixture_function(obj: Any) -> bool:
- obj = getattr(obj, "__func__", obj) # instance method maybe?
- return getattr(obj, "_force_asyncio_fixture", False)
-
-
-def _make_asyncio_fixture_function(obj: Any, loop_scope: _ScopeName | None) -> None:
- if hasattr(obj, "__func__"):
- # instance method, check the function object
- obj = obj.__func__
- obj._force_asyncio_fixture = True
- obj._loop_scope = loop_scope
-
-
-def _is_coroutine_or_asyncgen(obj: Any) -> bool:
- return inspect.iscoroutinefunction(obj) or inspect.isasyncgenfunction(obj)
-
-
-def _get_asyncio_mode(config: Config) -> Mode:
- val = config.getoption("asyncio_mode")
- if val is None:
- val = config.getini("asyncio_mode")
- try:
- return Mode(val)
- except ValueError as e:
- modes = ", ".join(m.value for m in Mode)
- raise pytest.UsageError(
- f"{val!r} is not a valid asyncio_mode. Valid modes: {modes}."
- ) from e
-
-
-_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET = """\
-The configuration option "asyncio_default_fixture_loop_scope" is unset.
-The event loop scope for asynchronous fixtures will default to the fixture caching \
-scope. Future versions of pytest-asyncio will default the loop scope for asynchronous \
-fixtures to function scope. Set the default fixture loop scope explicitly in order to \
-avoid unexpected behavior in the future. Valid fixture loop scopes are: \
-"function", "class", "module", "package", "session"
-"""
-
-
-def pytest_configure(config: Config) -> None:
- default_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
- if not default_loop_scope:
- warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
- config.addinivalue_line(
- "markers",
- "asyncio: "
- "mark the test as a coroutine, it will be "
- "run using an asyncio event loop",
- )
-
-
-@pytest.hookimpl(tryfirst=True)
-def pytest_report_header(config: Config) -> list[str]:
- """Add asyncio config to pytest header."""
- mode = _get_asyncio_mode(config)
- default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
- default_test_loop_scope = _get_default_test_loop_scope(config)
- header = [
- f"mode={mode}",
- f"asyncio_default_fixture_loop_scope={default_fixture_loop_scope}",
- f"asyncio_default_test_loop_scope={default_test_loop_scope}",
- ]
- return [
- "asyncio: " + ", ".join(header),
- ]
-
-
-def _preprocess_async_fixtures(
- collector: Collector,
- processed_fixturedefs: set[FixtureDef],
-) -> None:
- config = collector.config
- default_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
- asyncio_mode = _get_asyncio_mode(config)
- fixturemanager = config.pluginmanager.get_plugin("funcmanage")
- assert fixturemanager is not None
- for fixtures in fixturemanager._arg2fixturedefs.values():
- for fixturedef in fixtures:
- func = fixturedef.func
- if fixturedef in processed_fixturedefs or not _is_coroutine_or_asyncgen(
- func
- ):
- continue
- if asyncio_mode == Mode.STRICT and not _is_asyncio_fixture_function(func):
- # Ignore async fixtures without explicit asyncio mark in strict mode
- # This applies to pytest_trio fixtures, for example
- continue
- loop_scope = (
- getattr(func, "_loop_scope", None)
- or default_loop_scope
- or fixturedef.scope
- )
- _make_asyncio_fixture_function(func, loop_scope)
- if "request" not in fixturedef.argnames:
- fixturedef.argnames += ("request",)
- _synchronize_async_fixture(fixturedef)
- assert _is_asyncio_fixture_function(fixturedef.func)
- processed_fixturedefs.add(fixturedef)
-
-
-def _synchronize_async_fixture(fixturedef: FixtureDef) -> None:
- """Wraps the fixture function of an async fixture in a synchronous function."""
- if inspect.isasyncgenfunction(fixturedef.func):
- _wrap_asyncgen_fixture(fixturedef)
- elif inspect.iscoroutinefunction(fixturedef.func):
- _wrap_async_fixture(fixturedef)
-
-
-def _add_kwargs(
- func: Callable[..., Any],
- kwargs: dict[str, Any],
- request: FixtureRequest,
-) -> dict[str, Any]:
- sig = inspect.signature(func)
- ret = kwargs.copy()
- if "request" in sig.parameters:
- ret["request"] = request
- return ret
-
-
-def _perhaps_rebind_fixture_func(func: _T, instance: Any | None) -> _T:
- if instance is not None:
- # The fixture needs to be bound to the actual request.instance
- # so it is bound to the same object as the test method.
- unbound, cls = func, None
- try:
- unbound, cls = func.__func__, type(func.__self__) # type: ignore
- except AttributeError:
- pass
- # Only if the fixture was bound before to an instance of
- # the same type.
- if cls is not None and isinstance(instance, cls):
- func = unbound.__get__(instance) # type: ignore
- return func
-
-
-def _wrap_asyncgen_fixture(fixturedef: FixtureDef) -> None:
- fixture = fixturedef.func
-
- @functools.wraps(fixture)
- def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
- func = _perhaps_rebind_fixture_func(fixture, request.instance)
- event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
- request, func
- )
- event_loop = request.getfixturevalue(event_loop_fixture_id)
- kwargs.pop(event_loop_fixture_id, None)
- gen_obj = func(**_add_kwargs(func, kwargs, request))
-
- async def setup():
- res = await gen_obj.__anext__() # type: ignore[union-attr]
- return res
-
- context = contextvars.copy_context()
- setup_task = _create_task_in_context(event_loop, setup(), context)
- result = event_loop.run_until_complete(setup_task)
-
- reset_contextvars = _apply_contextvar_changes(context)
-
- def finalizer() -> None:
- """Yield again, to finalize."""
-
- async def async_finalizer() -> None:
- try:
- await gen_obj.__anext__() # type: ignore[union-attr]
- except StopAsyncIteration:
- pass
- else:
- msg = "Async generator fixture didn't stop."
- msg += "Yield only once."
- raise ValueError(msg)
-
- task = _create_task_in_context(event_loop, async_finalizer(), context)
- event_loop.run_until_complete(task)
- if reset_contextvars is not None:
- reset_contextvars()
-
- request.addfinalizer(finalizer)
- return result
-
- fixturedef.func = _asyncgen_fixture_wrapper # type: ignore[misc]
-
-
-def _wrap_async_fixture(fixturedef: FixtureDef) -> None:
- fixture = fixturedef.func
-
- @functools.wraps(fixture)
- def _async_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
- func = _perhaps_rebind_fixture_func(fixture, request.instance)
- event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
- request, func
- )
- event_loop = request.getfixturevalue(event_loop_fixture_id)
- kwargs.pop(event_loop_fixture_id, None)
-
- async def setup():
- res = await func(**_add_kwargs(func, kwargs, request))
- return res
-
- context = contextvars.copy_context()
- setup_task = _create_task_in_context(event_loop, setup(), context)
- result = event_loop.run_until_complete(setup_task)
-
- # Copy the context vars modified by the setup task into the current
- # context, and (if needed) add a finalizer to reset them.
- #
- # Note that this is slightly different from the behavior of a non-async
- # fixture, which would rely on the fixture author to add a finalizer
- # to reset the variables. In this case, the author of the fixture can't
- # write such a finalizer because they have no way to capture the Context
- # in which the setup function was run, so we need to do it for them.
- reset_contextvars = _apply_contextvar_changes(context)
- if reset_contextvars is not None:
- request.addfinalizer(reset_contextvars)
-
- return result
-
- fixturedef.func = _async_fixture_wrapper # type: ignore[misc]
-
-
-def _get_event_loop_fixture_id_for_async_fixture(
- request: FixtureRequest, func: Any
-) -> str:
- default_loop_scope = cast(
- _ScopeName, request.config.getini("asyncio_default_fixture_loop_scope")
- )
- loop_scope = (
- getattr(func, "_loop_scope", None) or default_loop_scope or request.scope
- )
- return f"_{loop_scope}_event_loop"
-
-
-def _create_task_in_context(
- loop: asyncio.AbstractEventLoop,
- coro: AbstractCoroutine[Any, Any, _T],
- context: contextvars.Context,
-) -> asyncio.Task[_T]:
- """
- Return an asyncio task that runs the coro in the specified context,
- if possible.
-
- This allows fixture setup and teardown to be run as separate asyncio tasks,
- while still being able to use context-manager idioms to maintain context
- variables and make those variables visible to test functions.
-
- This is only fully supported on Python 3.11 and newer, as it requires
- the API added for https://github.com/python/cpython/issues/91150.
- On earlier versions, the returned task will use the default context instead.
- """
- try:
- return loop.create_task(coro, context=context)
- except TypeError:
- return loop.create_task(coro)
-
-
-def _apply_contextvar_changes(
- context: contextvars.Context,
-) -> Callable[[], None] | None:
- """
- Copy contextvar changes from the given context to the current context.
-
- If any contextvars were modified by the fixture, return a finalizer that
- will restore them.
- """
- context_tokens = []
- for var in context:
- try:
- if var.get() is context.get(var):
- # This variable is not modified, so leave it as-is.
- continue
- except LookupError:
- # This variable isn't yet set in the current context at all.
- pass
- token = var.set(context.get(var))
- context_tokens.append((var, token))
-
- if not context_tokens:
- return None
-
- def restore_contextvars():
- while context_tokens:
- (var, token) = context_tokens.pop()
- var.reset(token)
-
- return restore_contextvars
-
-
-class PytestAsyncioFunction(Function):
- """Base class for all test functions managed by pytest-asyncio."""
-
- @classmethod
- def item_subclass_for(cls, item: Function, /) -> type[PytestAsyncioFunction] | None:
- """
- Returns a subclass of PytestAsyncioFunction if there is a specialized subclass
- for the specified function item.
-
- Return None if no specialized subclass exists for the specified item.
- """
- for subclass in cls.__subclasses__():
- if subclass._can_substitute(item):
- return subclass
- return None
-
- @classmethod
- def _from_function(cls, function: Function, /) -> Function:
- """
- Instantiates this specific PytestAsyncioFunction type from the specified
- Function item.
- """
- assert function.get_closest_marker("asyncio")
- subclass_instance = cls.from_parent(
- function.parent,
- name=function.name,
- callspec=getattr(function, "callspec", None),
- callobj=function.obj,
- fixtureinfo=function._fixtureinfo,
- keywords=function.keywords,
- originalname=function.originalname,
- )
- subclass_instance.own_markers = function.own_markers
- assert subclass_instance.own_markers == function.own_markers
- return subclass_instance
-
- @staticmethod
- def _can_substitute(item: Function) -> bool:
- """Returns whether the specified function can be replaced by this class"""
- raise NotImplementedError()
-
-
-class Coroutine(PytestAsyncioFunction):
- """Pytest item created by a coroutine"""
-
- @staticmethod
- def _can_substitute(item: Function) -> bool:
- func = item.obj
- return inspect.iscoroutinefunction(func)
-
- def runtest(self) -> None:
- self.obj = wrap_in_sync(
- # https://github.com/pytest-dev/pytest-asyncio/issues/596
- self.obj, # type: ignore[has-type]
- )
- super().runtest()
-
-
-class AsyncGenerator(PytestAsyncioFunction):
- """Pytest item created by an asynchronous generator"""
-
- @staticmethod
- def _can_substitute(item: Function) -> bool:
- func = item.obj
- return inspect.isasyncgenfunction(func)
-
- @classmethod
- def _from_function(cls, function: Function, /) -> Function:
- async_gen_item = super()._from_function(function)
- unsupported_item_type_message = (
- f"Tests based on asynchronous generators are not supported. "
- f"{function.name} will be ignored."
- )
- async_gen_item.warn(PytestCollectionWarning(unsupported_item_type_message))
- async_gen_item.add_marker(
- pytest.mark.xfail(run=False, reason=unsupported_item_type_message)
- )
- return async_gen_item
-
-
-class AsyncStaticMethod(PytestAsyncioFunction):
- """
- Pytest item that is a coroutine or an asynchronous generator
- decorated with staticmethod
- """
-
- @staticmethod
- def _can_substitute(item: Function) -> bool:
- func = item.obj
- return isinstance(func, staticmethod) and _is_coroutine_or_asyncgen(
- func.__func__
- )
-
- def runtest(self) -> None:
- self.obj = wrap_in_sync(
- # https://github.com/pytest-dev/pytest-asyncio/issues/596
- self.obj, # type: ignore[has-type]
- )
- super().runtest()
-
-
-class AsyncHypothesisTest(PytestAsyncioFunction):
- """
- Pytest item that is coroutine or an asynchronous generator decorated by
- @hypothesis.given.
- """
-
- @staticmethod
- def _can_substitute(item: Function) -> bool:
- func = item.obj
- return (
- getattr(func, "is_hypothesis_test", False) # type: ignore[return-value]
- and getattr(func, "hypothesis", None)
- and inspect.iscoroutinefunction(func.hypothesis.inner_test)
- )
-
- def runtest(self) -> None:
- self.obj.hypothesis.inner_test = wrap_in_sync(
- self.obj.hypothesis.inner_test,
- )
- super().runtest()
-
-
-_HOLDER: set[FixtureDef] = set()
-
-
-# The function name needs to start with "pytest_"
-# see https://github.com/pytest-dev/pytest/issues/11307
-@pytest.hookimpl(specname="pytest_pycollect_makeitem", tryfirst=True)
-def pytest_pycollect_makeitem_preprocess_async_fixtures(
- collector: pytest.Module | pytest.Class, name: str, obj: object
-) -> pytest.Item | pytest.Collector | list[pytest.Item | pytest.Collector] | None:
- """A pytest hook to collect asyncio coroutines."""
- if not collector.funcnamefilter(name):
- return None
- _preprocess_async_fixtures(collector, _HOLDER)
- return None
-
-
-# The function name needs to start with "pytest_"
-# see https://github.com/pytest-dev/pytest/issues/11307
-@pytest.hookimpl(specname="pytest_pycollect_makeitem", hookwrapper=True)
-def pytest_pycollect_makeitem_convert_async_functions_to_subclass(
- collector: pytest.Module | pytest.Class, name: str, obj: object
-) -> Generator[None, pluggy.Result, None]:
- """
- Converts coroutines and async generators collected as pytest.Functions
- to AsyncFunction items.
- """
- hook_result = yield
- try:
- node_or_list_of_nodes: (
- pytest.Item | pytest.Collector | list[pytest.Item | pytest.Collector] | None
- ) = hook_result.get_result()
- except BaseException as e:
- hook_result.force_exception(e)
- return
- if not node_or_list_of_nodes:
- return
- if isinstance(node_or_list_of_nodes, Sequence):
- node_iterator = iter(node_or_list_of_nodes)
- else:
- # Treat single node as a single-element iterable
- node_iterator = iter((node_or_list_of_nodes,))
- updated_node_collection = []
- for node in node_iterator:
- updated_item = node
- if isinstance(node, Function):
- specialized_item_class = PytestAsyncioFunction.item_subclass_for(node)
- if specialized_item_class:
- if _get_asyncio_mode(
- node.config
- ) == Mode.AUTO and not node.get_closest_marker("asyncio"):
- node.add_marker("asyncio")
- if node.get_closest_marker("asyncio"):
- updated_item = specialized_item_class._from_function(node)
- updated_node_collection.append(updated_item)
- hook_result.force_result(updated_node_collection)
-
-
-@contextlib.contextmanager
-def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]:
- old_loop_policy = _get_event_loop_policy()
- try:
- old_loop = _get_event_loop_no_warn()
- except RuntimeError:
- old_loop = None
- _set_event_loop_policy(policy)
- try:
- yield
- finally:
- _set_event_loop_policy(old_loop_policy)
- _set_event_loop(old_loop)
-
-
-@pytest.hookimpl(tryfirst=True)
-def pytest_generate_tests(metafunc: Metafunc) -> None:
- marker = metafunc.definition.get_closest_marker("asyncio")
- if not marker:
- return
- default_loop_scope = _get_default_test_loop_scope(metafunc.config)
- loop_scope = _get_marked_loop_scope(marker, default_loop_scope)
- event_loop_fixture_id = f"_{loop_scope}_event_loop"
- # This specific fixture name may already be in metafunc.argnames, if this
- # test indirectly depends on the fixture. For example, this is the case
- # when the test depends on an async fixture, both of which share the same
- # event loop fixture mark.
- if event_loop_fixture_id in metafunc.fixturenames:
- return
- fixturemanager = metafunc.config.pluginmanager.get_plugin("funcmanage")
- assert fixturemanager is not None
- # Add the scoped event loop fixture to Metafunc's list of fixture names and
- # fixturedefs and leave the actual parametrization to pytest
- # The fixture needs to be appended to avoid messing up the fixture evaluation
- # order
- metafunc.fixturenames.append(event_loop_fixture_id)
- metafunc._arg2fixturedefs[event_loop_fixture_id] = fixturemanager._arg2fixturedefs[
- event_loop_fixture_id
- ]
-
-
-def _get_event_loop_policy() -> AbstractEventLoopPolicy:
- with warnings.catch_warnings():
- warnings.simplefilter("ignore", DeprecationWarning)
- return asyncio.get_event_loop_policy()
-
-
-def _set_event_loop_policy(policy: AbstractEventLoopPolicy) -> None:
- with warnings.catch_warnings():
- warnings.simplefilter("ignore", DeprecationWarning)
- asyncio.set_event_loop_policy(policy)
-
-
-def _get_event_loop_no_warn(
- policy: AbstractEventLoopPolicy | None = None,
-) -> asyncio.AbstractEventLoop:
- with warnings.catch_warnings():
- warnings.simplefilter("ignore", DeprecationWarning)
- if policy is not None:
- return policy.get_event_loop()
- else:
- return asyncio.get_event_loop()
-
-
-def _set_event_loop(loop: AbstractEventLoop | None) -> None:
- with warnings.catch_warnings():
- warnings.simplefilter("ignore", DeprecationWarning)
- asyncio.set_event_loop(loop)
-
-
-@pytest.hookimpl(tryfirst=True, hookwrapper=True)
-def pytest_pyfunc_call(pyfuncitem: Function) -> object | None:
- """
- Pytest hook called before a test case is run.
-
- Wraps marked tests in a synchronous function
- where the wrapped test coroutine is executed in an event loop.
- """
- if pyfuncitem.get_closest_marker("asyncio") is not None:
- if isinstance(pyfuncitem, PytestAsyncioFunction):
- asyncio_mode = _get_asyncio_mode(pyfuncitem.config)
- for fixname, fixtures in pyfuncitem._fixtureinfo.name2fixturedefs.items():
- # name2fixturedefs is a dict between fixture name and a list of matching
- # fixturedefs. The last entry in the list is closest and the one used.
- func = fixtures[-1].func
- if (
- asyncio_mode == Mode.STRICT
- and _is_coroutine_or_asyncgen(func)
- and not _is_asyncio_fixture_function(func)
- ):
- warnings.warn(
- PytestDeprecationWarning(
- f"asyncio test {pyfuncitem.name!r} requested async "
- "@pytest.fixture "
- f"{fixname!r} in strict mode. "
- "You might want to use @pytest_asyncio.fixture or switch "
- "to auto mode. "
- "This will become an error in future versions of "
- "flake8-asyncio."
- ),
- stacklevel=1,
- )
- # no stacklevel points at the users code, so we set stacklevel=1
- # so it at least indicates that it's the plugin complaining.
- # Pytest gives the test file & name in the warnings summary at least
-
- else:
- pyfuncitem.warn(
- pytest.PytestWarning(
- f"The test {pyfuncitem} is marked with '@pytest.mark.asyncio' "
- "but it is not an async function. "
- "Please remove the asyncio mark. "
- "If the test is not marked explicitly, "
- "check for global marks applied via 'pytestmark'."
- )
- )
- yield
- return None
-
-
-def wrap_in_sync(
- func: Callable[..., Awaitable[Any]],
-):
- """
- Return a sync wrapper around an async function executing it in the
- current event loop.
- """
- # if the function is already wrapped, we rewrap using the original one
- # not using __wrapped__ because the original function may already be
- # a wrapped one
- raw_func = getattr(func, "_raw_test_func", None)
- if raw_func is not None:
- func = raw_func
-
- @functools.wraps(func)
- def inner(*args, **kwargs):
- coro = func(*args, **kwargs)
- _loop = _get_event_loop_no_warn()
- task = asyncio.ensure_future(coro, loop=_loop)
- try:
- _loop.run_until_complete(task)
- except BaseException:
- # run_until_complete doesn't get the result from exceptions
- # that are not subclasses of `Exception`. Consume all
- # exceptions to prevent asyncio's warning from logging.
- if task.done() and not task.cancelled():
- task.exception()
- raise
-
- inner._raw_test_func = func # type: ignore[attr-defined]
- return inner
-
-
-def pytest_runtest_setup(item: pytest.Item) -> None:
- marker = item.get_closest_marker("asyncio")
- if marker is None:
- return
- default_loop_scope = _get_default_test_loop_scope(item.config)
- loop_scope = _get_marked_loop_scope(marker, default_loop_scope)
- event_loop_fixture_id = f"_{loop_scope}_event_loop"
- fixturenames = item.fixturenames # type: ignore[attr-defined]
- if event_loop_fixture_id not in fixturenames:
- fixturenames.append(event_loop_fixture_id)
- obj = getattr(item, "obj", None)
- if not getattr(obj, "hypothesis", False) and getattr(
- obj, "is_hypothesis_test", False
- ):
- pytest.fail(
- f"test function `{item!r}` is using Hypothesis, but pytest-asyncio "
- "only works with Hypothesis 3.64.0 or later."
- )
-
-
-_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR = """\
-An asyncio pytest marker defines both "scope" and "loop_scope", \
-but it should only use "loop_scope".
-"""
-
-_MARKER_SCOPE_KWARG_DEPRECATION_WARNING = """\
-The "scope" keyword argument to the asyncio marker has been deprecated. \
-Please use the "loop_scope" argument instead.
-"""
-
-
-def _get_marked_loop_scope(
- asyncio_marker: Mark, default_loop_scope: _ScopeName
-) -> _ScopeName:
- assert asyncio_marker.name == "asyncio"
- if asyncio_marker.args or (
- asyncio_marker.kwargs and set(asyncio_marker.kwargs) - {"loop_scope", "scope"}
- ):
- raise ValueError("mark.asyncio accepts only a keyword argument 'loop_scope'.")
- if "scope" in asyncio_marker.kwargs:
- if "loop_scope" in asyncio_marker.kwargs:
- raise pytest.UsageError(_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR)
- warnings.warn(PytestDeprecationWarning(_MARKER_SCOPE_KWARG_DEPRECATION_WARNING))
- scope = asyncio_marker.kwargs.get("loop_scope") or asyncio_marker.kwargs.get(
- "scope"
- )
- if scope is None:
- scope = default_loop_scope
- assert scope in {"function", "class", "module", "package", "session"}
- return scope
-
-
-def _get_default_test_loop_scope(config: Config) -> _ScopeName:
- return config.getini("asyncio_default_test_loop_scope")
-
-
-def _create_scoped_event_loop_fixture(scope: _ScopeName) -> Callable:
- @pytest.fixture(
- scope=scope,
- name=f"_{scope}_event_loop",
- )
- def _scoped_event_loop(
- *args, # Function needs to accept "cls" when collected by pytest.Class
- event_loop_policy,
- ) -> Iterator[asyncio.AbstractEventLoop]:
- new_loop_policy = event_loop_policy
- with (
- _temporary_event_loop_policy(new_loop_policy),
- _provide_event_loop() as loop,
- ):
- _set_event_loop(loop)
- yield loop
-
- return _scoped_event_loop
-
-
-for scope in Scope:
- globals()[f"_{scope.value}_event_loop"] = _create_scoped_event_loop_fixture(
- scope.value
- )
-
-
-@contextlib.contextmanager
-def _provide_event_loop() -> Iterator[asyncio.AbstractEventLoop]:
- policy = _get_event_loop_policy()
- loop = policy.new_event_loop()
- try:
- yield loop
- finally:
- # cleanup the event loop if it hasn't been cleaned up already
- if not loop.is_closed():
- try:
- loop.run_until_complete(loop.shutdown_asyncgens())
- except Exception as e:
- warnings.warn(f"Error cleaning up asyncio loop: {e}", RuntimeWarning)
- finally:
- loop.close()
-
-
-@pytest.fixture(scope="session", autouse=True)
-def event_loop_policy() -> AbstractEventLoopPolicy:
- """Return an instance of the policy used to create asyncio event loops."""
- return _get_event_loop_policy()
-
-
-def is_async_test(item: Item) -> bool:
- """Returns whether a test item is a pytest-asyncio test"""
- return isinstance(item, PytestAsyncioFunction)
-
-
-def _unused_port(socket_type: int) -> int:
- """Find an unused localhost port from 1024-65535 and return it."""
- with contextlib.closing(socket.socket(type=socket_type)) as sock:
- sock.bind(("127.0.0.1", 0))
- return sock.getsockname()[1]
-
-
-@pytest.fixture
-def unused_tcp_port() -> int:
- return _unused_port(socket.SOCK_STREAM)
-
-
-@pytest.fixture
-def unused_udp_port() -> int:
- return _unused_port(socket.SOCK_DGRAM)
-
-
-@pytest.fixture(scope="session")
-def unused_tcp_port_factory() -> Callable[[], int]:
- """A factory function, producing different unused TCP ports."""
- produced = set()
-
- def factory():
- """Return an unused port."""
- port = _unused_port(socket.SOCK_STREAM)
-
- while port in produced:
- port = _unused_port(socket.SOCK_STREAM)
-
- produced.add(port)
-
- return port
-
- return factory
-
-
-@pytest.fixture(scope="session")
-def unused_udp_port_factory() -> Callable[[], int]:
- """A factory function, producing different unused UDP ports."""
- produced = set()
-
- def factory():
- """Return an unused port."""
- port = _unused_port(socket.SOCK_DGRAM)
-
- while port in produced:
- port = _unused_port(socket.SOCK_DGRAM)
-
- produced.add(port)
-
- return port
-
- return factory
projects / fastapi-vs-flask / .git / commitdiff