diff --git a/airflow-core/src/airflow/serialization/enums.py b/airflow-core/src/airflow/serialization/enums.py index a1f16ac9bfb..ee0686ae348 100644 --- a/airflow-core/src/airflow/serialization/enums.py +++ b/airflow-core/src/airflow/serialization/enums.py @@ -36,11 +36,10 @@ def stringify_encoding_keys(d: Any) -> Any: """ Convert BaseSerialization Encoding enum keys to their string values recursively. - Python 3.10 compatibility: str(Encoding.TYPE) returns "Encoding.TYPE" on 3.10 - instead of "__type__" (3.10 is still the default CI target). serde.serialize - uses str(k) for dict keys, so without this conversion the encrypted blob ends up - with "Encoding.TYPE" keys that neither serde._convert nor the BaseSerialization - fallback can read back. + str(Encoding.TYPE) returns "Encoding.TYPE" instead of "__type" for a ``(str, Enum)`` + mixin. serde.serialize uses str(k) for dict keys, so without this conversion the + encrypted blob ends up with "Encoding.TYPE" keys that neither serde._convert nor the + BaseSerialization fallback can read back. """ if isinstance(d, dict): return { diff --git a/airflow-core/src/airflow/serialization/stub_arg_bindings.py b/airflow-core/src/airflow/serialization/stub_arg_bindings.py index 864848c49ae..6746a83fd58 100644 --- a/airflow-core/src/airflow/serialization/stub_arg_bindings.py +++ b/airflow-core/src/airflow/serialization/stub_arg_bindings.py @@ -75,9 +75,6 @@ def _normalize_temporal_annotation(annotation: Any) -> Any: Applied recursively through unions and containers, and only as a retry when direct schema generation fails, so temporal types carrying their own pydantic schema keep it. """ - # Parametrized generics must be detected before the plain-class branch: on Python - # 3.10, isinstance(list[X], type) is True and issubclass silently consults the - # origin, so the class branch would return list[X] unnormalized. origin = typing.get_origin(annotation) args = typing.get_args(annotation) if origin is not None and args: diff --git a/airflow-core/src/airflow/typing_compat.py b/airflow-core/src/airflow/typing_compat.py index 1f9fd0b217f..576da102dcc 100644 --- a/airflow-core/src/airflow/typing_compat.py +++ b/airflow-core/src/airflow/typing_compat.py @@ -21,13 +21,5 @@ from __future__ import annotations __all__ = ["Literal", "ParamSpec", "Self", "TypeAlias", "TypeGuard", "Unpack", "assert_never"] -import sys - # Keeping this for backwards-compat with old providers -from typing import Literal, ParamSpec, TypeAlias, TypeGuard - -if sys.version_info >= (3, 11): - from typing import Self, Unpack, assert_never -else: - # TODO: Remove once Python 3.10 support is dropped (EOL 2026) - from typing_extensions import Self, Unpack, assert_never +from typing import Literal, ParamSpec, Self, TypeAlias, TypeGuard, Unpack, assert_never diff --git a/airflow-core/tests/unit/dag_processing/test_dagbag.py b/airflow-core/tests/unit/dag_processing/test_dagbag.py index 4c0f744fa1e..dc02e9ea085 100644 --- a/airflow-core/tests/unit/dag_processing/test_dagbag.py +++ b/airflow-core/tests/unit/dag_processing/test_dagbag.py @@ -27,7 +27,7 @@ import textwrap import warnings import zipfile from copy import deepcopy -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from pathlib import Path from unittest import mock from unittest.mock import patch @@ -100,7 +100,6 @@ def standard_example_dags_folder() -> Path: return _standard_example_dags_folder() -PY311 = sys.version_info >= (3, 11) PY313 = sys.version_info >= (3, 13) # Include the words "airflow" and "dag" in the file contents, @@ -1024,7 +1023,7 @@ class TestDagBag: fileloc = str(standard_example_dags_folder / "example_bash_operator.py") mock_dagmodel.return_value = DagModel() - mock_dagmodel.return_value.last_expired = datetime.max.replace(tzinfo=timezone.utc) + mock_dagmodel.return_value.last_expired = datetime.max.replace(tzinfo=UTC) mock_dagmodel.return_value.fileloc = fileloc class _TestDagBag(DagBag): @@ -1057,7 +1056,7 @@ class TestDagBag: fileloc = os.path.realpath(os.path.join(test_zip_path, "test_zip.py")) mock_dagmodel.return_value = DagModel() - mock_dagmodel.return_value.last_expired = datetime.max.replace(tzinfo=timezone.utc) + mock_dagmodel.return_value.last_expired = datetime.max.replace(tzinfo=UTC) mock_dagmodel.return_value.fileloc = fileloc processed: list[str] = [] @@ -1180,7 +1179,7 @@ with airflow.DAG( @staticmethod def _make_test_traceback(unparseable_filename: str, depth=None) -> str: - python_311_marker = " ^^^^^^^^^^^\n" if PY311 else "" + python_311_marker = " ^^^^^^^^^^^\n" python_313_marker = [" ~~~~~~~~~^^\n"] if PY313 else [] frames = ( f' File "{unparseable_filename}", line 3, in \n something()\n', diff --git a/dev/breeze/src/airflow_breeze/commands/registry_commands.py b/dev/breeze/src/airflow_breeze/commands/registry_commands.py index 97164351518..e49ecfb275d 100644 --- a/dev/breeze/src/airflow_breeze/commands/registry_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/registry_commands.py @@ -181,10 +181,7 @@ def _find_provider_yaml(provider_id: str) -> Path: def _read_provider_yaml_info(provider_id: str) -> tuple[str, list[str]]: """Read package name from provider.yaml and extras from pyproject.toml.""" - try: - import tomllib - except ImportError: - import tomli as tomllib # type: ignore[no-redef] + import tomllib provider_yaml_path = _find_provider_yaml(provider_id) with open(provider_yaml_path) as f: diff --git a/dev/breeze/src/airflow_breeze/utils/packages.py b/dev/breeze/src/airflow_breeze/utils/packages.py index 421a13d2b65..74bd35f827c 100644 --- a/dev/breeze/src/airflow_breeze/utils/packages.py +++ b/dev/breeze/src/airflow_breeze/utils/packages.py @@ -531,10 +531,8 @@ def get_provider_yaml(provider_id: str) -> Path: def load_pyproject_toml(pyproject_toml_file_path: Path) -> dict[str, Any]: - try: - import tomllib - except ImportError: - import tomli as tomllib # type: ignore[no-redef] + import tomllib + toml_content = pyproject_toml_file_path.read_text() syntax = Syntax(toml_content, "toml", theme="ansi_dark", line_numbers=True) try: diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py b/dev/breeze/src/airflow_breeze/utils/selective_checks.py index 5d4f268cc59..38c0f078057 100644 --- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py +++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py @@ -1823,10 +1823,7 @@ class SelectiveChecks: f"Could not get pyproject.toml from {self._commit_ref}^[/]" ) return False - try: - import tomllib - except ImportError: - import tomli as tomllib # type: ignore[no-redef] + import tomllib self._new_toml = tomllib.loads(new_result.stdout) self._old_toml = tomllib.loads(old_result.stdout) @@ -2266,10 +2263,7 @@ class SelectiveChecks: if not pyproject_files or not self._github_event == GithubEvents.PULL_REQUEST: return False - try: - import tomllib - except ImportError: - import tomli as tomllib # type: ignore[no-redef] + import tomllib violations = [] for pyproject_file in pyproject_files: diff --git a/devel-common/src/sphinx_exts/docs_build/package_filter.py b/devel-common/src/sphinx_exts/docs_build/package_filter.py index 3cd83951be8..52823161cfe 100644 --- a/devel-common/src/sphinx_exts/docs_build/package_filter.py +++ b/devel-common/src/sphinx_exts/docs_build/package_filter.py @@ -61,10 +61,8 @@ def find_packages_to_build(available_packages: list[str], package_filters: list[ elif folder_name == "task-sdk": package_name = "task-sdk" else: - try: - import tomllib - except ImportError: - import tomli as tomllib # type: ignore[no-redef] + import tomllib + read_toml = tomllib.loads(pyproject_toml_path.read_text()) package_name = read_toml["project"]["name"] if package_name == "apache-airflow": diff --git a/providers/common/ai/src/airflow/providers/common/ai/utils/usage.py b/providers/common/ai/src/airflow/providers/common/ai/utils/usage.py index 400edb6b753..d45f59d06a4 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/utils/usage.py +++ b/providers/common/ai/src/airflow/providers/common/ai/utils/usage.py @@ -55,11 +55,8 @@ def _resolve_field_type(field: str, hint: Any) -> type: raise TypeError(f"UsageLimits.{field} has an unsupported annotation {hint!r}") if typing.get_origin(resolved) is not None: # ``list[int] | None`` passes the top-level check (its origin is a union) - # and then reduces to ``list[int]``, which is a ``type`` instance on - # Python 3.10 -- only 3.11+ made ``isinstance(list[int], type)`` False - # (gh-101162). Without this the annotation the comment above names as - # rejected would resolve to a container type on the oldest supported - # Python, and a rendered value would reach ``UsageLimits`` unchecked. + # and then reduces to ``list[int]``; reject it here so the error names the + # unsupported annotation instead of a bare non-type. raise TypeError(f"UsageLimits.{field} has an unsupported annotation {hint!r}") if not isinstance(resolved, type): raise TypeError(f"UsageLimits.{field} resolved to a non-type {resolved!r}") diff --git a/providers/common/ai/tests/unit/common/ai/utils/test_toolset_base.py b/providers/common/ai/tests/unit/common/ai/utils/test_toolset_base.py index 7121cf85175..f95afa9e42e 100644 --- a/providers/common/ai/tests/unit/common/ai/utils/test_toolset_base.py +++ b/providers/common/ai/tests/unit/common/ai/utils/test_toolset_base.py @@ -17,7 +17,6 @@ from __future__ import annotations import asyncio -import sys import threading import time import traceback @@ -162,7 +161,6 @@ class TestMasking: assert caught.value.metadata == {"reason": "***"} assert "failed" not in caplog.text - @pytest.mark.skipif(sys.version_info < (3, 11), reason="ExceptionGroup is built in from Python 3.11") def test_masks_the_exceptions_inside_an_exception_group(self, registered_secret): cause = ConnectionError(f"socket closed by {registered_secret}") inner = ValueError(f"login failed for {registered_secret}") diff --git a/providers/common/ai/tests/unit/common/ai/utils/test_usage.py b/providers/common/ai/tests/unit/common/ai/utils/test_usage.py index 13a55cd9bb0..a56354cc126 100644 --- a/providers/common/ai/tests/unit/common/ai/utils/test_usage.py +++ b/providers/common/ai/tests/unit/common/ai/utils/test_usage.py @@ -494,12 +494,10 @@ class TestResolveFieldType: ], ) def test_optional_generic_raises(self, hint): - """A generic wrapped in ``| None`` must be rejected on every supported Python. + """A generic wrapped in ``| None`` must be rejected as an unsupported annotation. The union passes the top-level origin check and reduces to ``list[int]``, - which is a ``type`` instance on 3.10 and only stopped being one in 3.11 - (gh-101162). Without checking the reduced member this resolved to a - container type on the oldest supported Python while passing on 3.12. + which must be rejected explicitly rather than as a bare non-type. """ with pytest.raises(TypeError, match="unsupported annotation"): _resolve_field_type("some_field", hint) diff --git a/providers/standard/tests/unit/standard/decorators/test_python.py b/providers/standard/tests/unit/standard/decorators/test_python.py index 252ecc5c1b0..bd942a670db 100644 --- a/providers/standard/tests/unit/standard/decorators/test_python.py +++ b/providers/standard/tests/unit/standard/decorators/test_python.py @@ -16,7 +16,6 @@ # specific language governing permissions and limitations # under the License. -import sys import typing from collections import namedtuple from datetime import date @@ -81,8 +80,6 @@ if typing.TYPE_CHECKING: pytestmark = pytest.mark.db_test DEFAULT_DATE = timezone.datetime(2016, 1, 1) -PY38 = sys.version_info >= (3, 8) -PY311 = sys.version_info >= (3, 11) class TestAirflowTaskDecorator(BasePythonTest): @@ -155,16 +152,10 @@ class TestAirflowTaskDecorator(BasePythonTest): assert t1().operator.multiple_outputs is True - # We do not enable `from __future__ import annotations` for particular this test module, - # that mean `str | None` annotation would raise TypeError in Python 3.9 and below - @pytest.mark.skipif(sys.version_info < (3, 10), reason="PEP 604 is implemented in Python 3.10") def test_infer_multiple_outputs_pep_604_union_type(self): @task_decorator def t1() -> str | None: - # Before PEP 604 which are implemented in Python 3.10 `str | None` - # returns `types.UnionType` which are class and could be check in `issubclass()`. - # However in Python 3.10+ this construction returns object `typing.Union` - # which can not be used in `issubclass()` + # `str | None` evaluates to `types.UnionType`, which cannot be used in `issubclass()`. return "foo" assert t1().operator.multiple_outputs is False diff --git a/scripts/tools/initialize_virtualenv.py b/scripts/tools/initialize_virtualenv.py index a6edf0268dc..ae67ffe4070 100755 --- a/scripts/tools/initialize_virtualenv.py +++ b/scripts/tools/initialize_virtualenv.py @@ -59,10 +59,8 @@ def get_dependency_groups(pyproject_toml_path: Path) -> list[str]: """ Get the dependency groups from pyproject.toml """ - try: - import tomllib - except ImportError: - import tomli as tomllib # type: ignore[no-redef] + import tomllib + airflow_core_toml_dict = tomllib.loads(pyproject_toml_path.read_text()) return airflow_core_toml_dict["dependency-groups"].keys() diff --git a/shared/dagnode/src/airflow_shared/dagnode/node.py b/shared/dagnode/src/airflow_shared/dagnode/node.py index 651547a183b..9bc3b91f60a 100644 --- a/shared/dagnode/src/airflow_shared/dagnode/node.py +++ b/shared/dagnode/src/airflow_shared/dagnode/node.py @@ -22,14 +22,8 @@ from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar import structlog if TYPE_CHECKING: - import sys from collections.abc import Collection, Iterable - - # Replicate `airflow.typing_compat.Self` to avoid illegal imports - if sys.version_info >= (3, 11): - from typing import Self - else: - from typing_extensions import Self + from typing import Self from ..logging.types import Logger diff --git a/shared/logging/tests/logging/test_structlog.py b/shared/logging/tests/logging/test_structlog.py index cf63d2ed778..bb2533c5ea7 100644 --- a/shared/logging/tests/logging/test_structlog.py +++ b/shared/logging/tests/logging/test_structlog.py @@ -24,7 +24,7 @@ import logging import os import sys import textwrap -from datetime import datetime, timezone +from datetime import UTC, datetime from pathlib import Path from unittest import mock @@ -43,12 +43,10 @@ from airflow_shared.logging.structlog import ( # We avoid the caplog fixture for most tests here; the main purpose of this file is to capture the # _rendered_ output of the tests to make sure it is correct. -PY_3_11 = sys.version_info >= (3, 11) - @pytest.fixture(autouse=True) def set_time(time_machine): - time_machine.move_to(datetime(1985, 10, 26, microsecond=1, tzinfo=timezone.utc), tick=False) + time_machine.move_to(datetime(1985, 10, 26, microsecond=1, tzinfo=UTC), tick=False) @pytest.fixture @@ -368,8 +366,7 @@ def test_precent_fmt_exc(structlog_config, get_logger, monkeypatch): File "{__file__}", line {lineno}, in test_precent_fmt_exc 1 / 0 """) - if PY_3_11: - expected += " ~~^~~\n" + expected += " ~~^~~\n" expected += "ZeroDivisionError: division by zero\n" assert written == expected diff --git a/shared/observability/src/airflow_shared/observability/__init__.py b/shared/observability/src/airflow_shared/observability/__init__.py index f0f5cff5b87..3222bce6d93 100644 --- a/shared/observability/src/airflow_shared/observability/__init__.py +++ b/shared/observability/src/airflow_shared/observability/__init__.py @@ -16,11 +16,6 @@ # under the License. from __future__ import annotations -import sys - -if sys.version_info >= (3, 11): - from typing import Self -else: - from typing_extensions import Self +from typing import Self __all__ = ["Self"] diff --git a/task-sdk/dev/generate_task_sdk_models.py b/task-sdk/dev/generate_task_sdk_models.py index bb19fffc449..af0bc10245e 100644 --- a/task-sdk/dev/generate_task_sdk_models.py +++ b/task-sdk/dev/generate_task_sdk_models.py @@ -47,10 +47,7 @@ task_sdk_root = Path(__file__).parents[1] def load_config(): - try: - from tomllib import load as load_tomllib - except ImportError: - from tomli import load as load_tomllib + from tomllib import load as load_tomllib pyproject = AIRFLOW_TASK_SDK_ROOT_PATH / "pyproject.toml" # Simulate what `datamodel-code-generator` does on the CLI diff --git a/task-sdk/src/airflow/sdk/bases/operator.py b/task-sdk/src/airflow/sdk/bases/operator.py index 7634443b119..536b4545d17 100644 --- a/task-sdk/src/airflow/sdk/bases/operator.py +++ b/task-sdk/src/airflow/sdk/bases/operator.py @@ -85,9 +85,9 @@ T = TypeVar("T", bound=FunctionType) if TYPE_CHECKING: from types import ClassMethodDescriptorType + from typing import Self import jinja2 - from typing_extensions import Self from airflow.sdk.api.datamodels._generated import DagAttributeTypes from airflow.sdk.bases.operatorlink import BaseOperatorLink @@ -200,25 +200,6 @@ def coerce_resources(resources: dict[str, Any] | None) -> Resources | None: return Resources(**resources) -def _cancel_all_tasks(loop: AbstractEventLoop) -> None: - """Cancel every task still pending on ``loop`` and let it run its cleanup, as ``asyncio.run()`` does.""" - to_cancel = asyncio.all_tasks(loop) - if not to_cancel: - return - for task in to_cancel: - task.cancel() - loop.run_until_complete(asyncio.gather(*to_cancel, return_exceptions=True)) - for task in to_cancel: - if not task.cancelled() and task.exception() is not None: - loop.call_exception_handler( - { - "message": "unhandled exception during event_loop() shutdown", - "exception": task.exception(), - "task": task, - } - ) - - @contextlib.contextmanager def event_loop() -> Generator[AbstractEventLoop]: """ @@ -228,10 +209,9 @@ def event_loop() -> Generator[AbstractEventLoop]: outlives any number of ``run_until_complete()`` calls: tasks created on it in one call are still there for the next, which is what code that pauses and resumes a loop between batches of work needs. - On Python 3.11+ this is :class:`asyncio.Runner`; Python 3.10 has no ``Runner`` and gets a fallback that - replicates ``Runner.close()``. In both cases the loop is set as the current loop of the thread while the - block runs and unset afterwards, leftover tasks are cancelled, async generators and the default executor - are shut down and the loop is closed. Nothing goes through ``asyncio.get_event_loop()``, which since + This is :class:`asyncio.Runner`: the loop is set as the current loop of the thread while the block runs + and unset afterwards, leftover tasks are cancelled, async generators and the default executor are shut + down and the loop is closed. Nothing goes through ``asyncio.get_event_loop()``, which since Python 3.12 warns and since 3.14 raises when no loop is set. Entering the block from a running loop is refused: a loop cannot drive another loop on the same thread. """ @@ -242,24 +222,8 @@ def event_loop() -> Generator[AbstractEventLoop]: else: raise RuntimeError("event_loop() cannot be used from a running event loop") - if Runner := getattr(asyncio, "Runner", None): - with Runner() as runner: - yield runner.get_loop() - return - - # Python 3.10 has no asyncio.Runner: replicate what Runner.close() does. Drop with Python 3.10 support. - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - yield loop - finally: - try: - _cancel_all_tasks(loop) - loop.run_until_complete(loop.shutdown_asyncgens()) - loop.run_until_complete(loop.shutdown_default_executor()) - finally: - asyncio.set_event_loop(None) - loop.close() + with asyncio.Runner() as runner: + yield runner.get_loop() class _PartialDescriptor: diff --git a/task-sdk/src/airflow/sdk/serde/__init__.py b/task-sdk/src/airflow/sdk/serde/__init__.py index 1b0e06853a1..b9fb4cc25f6 100644 --- a/task-sdk/src/airflow/sdk/serde/__init__.py +++ b/task-sdk/src/airflow/sdk/serde/__init__.py @@ -139,9 +139,6 @@ def iter_pydantic_models(annotation: Any) -> Iterator[type]: stack: list[Any] = [annotation] while stack: tp = stack.pop() - # ``list[A]`` answers ``True`` to ``isinstance(tp, type)`` on 3.10+ yet - # carries a non-None ``get_origin``; recurse into its args first so the - # container itself is not mistaken for a leaf type. origin = get_origin(tp) if origin is not None: stack.extend(get_args(tp)) diff --git a/task-sdk/tests/task_sdk/api/test_client.py b/task-sdk/tests/task_sdk/api/test_client.py index 6a80ebb2a96..92bb781abd2 100644 --- a/task-sdk/tests/task_sdk/api/test_client.py +++ b/task-sdk/tests/task_sdk/api/test_client.py @@ -19,8 +19,7 @@ from __future__ import annotations import json import pickle -import sys -from datetime import datetime, timezone as dt_timezone +from datetime import UTC, datetime from typing import TYPE_CHECKING from unittest import mock @@ -198,7 +197,6 @@ class TestClient: assert unpickled.response.status_code == 404 assert unpickled.request.url == "http://error" - @pytest.mark.skipif(sys.version_info < (3, 11), reason="Exception notes (PEP 678) require Python 3.11") def test_server_error_detail_added_as_note(self): """Notes survive uncaught propagation, handled sites still log detail directly.""" responses = [httpx.Response(404, json={"detail": {"message": "Invalid input"}})] @@ -1392,7 +1390,7 @@ class TestAssetEventOperations: result = client.asset_events.get( name="this_asset", partition_key_regexp_pattern=r"^us\|2024-.*", - after=datetime(2023, 6, 1, tzinfo=dt_timezone.utc), + after=datetime(2023, 6, 1, tzinfo=UTC), limit=5, ascending=False, ) @@ -2006,7 +2004,7 @@ class TestDagsOperations: relative_fileloc="dags/example.py", owners="owner_1", tags=["a_tag", "z_tag"], - next_dagrun=datetime(2026, 4, 13, tzinfo=dt_timezone.utc), + next_dagrun=datetime(2026, 4, 13, tzinfo=UTC), ) def test_get_url_quotes_dag_id_as_single_path_segment(self): @@ -2129,7 +2127,7 @@ class TestTaskStateOperations: assert result.error == ErrorType.TASK_STORE_NOT_FOUND def test_set_success(self): - expires = datetime(2026, 6, 13, 12, 0, 0, tzinfo=dt_timezone.utc) + expires = datetime(2026, 6, 13, 12, 0, 0, tzinfo=UTC) def handle_request(request: httpx.Request) -> httpx.Response: assert request.method == "PUT" @@ -2160,7 +2158,7 @@ class TestTaskStateOperations: def test_set_with_expires_at_sends_field(self): """expires_at is forwarded as an ISO datetime string in the request body.""" - expires = datetime(2026, 5, 21, 12, 0, 0, tzinfo=dt_timezone.utc) + expires = datetime(2026, 5, 21, 12, 0, 0, tzinfo=UTC) def handle_request(request: httpx.Request) -> httpx.Response: body = json.loads(request.content) diff --git a/task-sdk/tests/task_sdk/bases/test_operator.py b/task-sdk/tests/task_sdk/bases/test_operator.py index 9041e404b12..d2060362f2b 100644 --- a/task-sdk/tests/task_sdk/bases/test_operator.py +++ b/task-sdk/tests/task_sdk/bases/test_operator.py @@ -22,7 +22,7 @@ import copy import logging import uuid import warnings -from datetime import date, datetime, timedelta, timezone +from datetime import UTC, date, datetime, timedelta from typing import NamedTuple from unittest import mock @@ -46,7 +46,7 @@ from airflow.sdk.definitions.param import ParamsDict from airflow.sdk.definitions.template import literal from airflow.triggers.base import StartTriggerArgs -DEFAULT_DATE = datetime(2016, 1, 1, tzinfo=timezone.utc) +DEFAULT_DATE = datetime(2016, 1, 1, tzinfo=UTC) class ClassWithCustomAttributes: @@ -1142,21 +1142,6 @@ def test_partial_default_args(): assert op.queue == "THIS" -@pytest.fixture(params=["asyncio.Runner", "fallback"]) -def event_loop_runner(request, monkeypatch): - """ - Run a test against both ways ``event_loop()`` can own its loop. - - ``asyncio.Runner`` exists from Python 3.11; the fallback replicates it for Python 3.10 and is - exercised on every Python by hiding ``asyncio.Runner`` for the test. - """ - if request.param == "fallback": - monkeypatch.delattr(asyncio, "Runner", raising=False) - elif not hasattr(asyncio, "Runner"): - pytest.skip("asyncio.Runner needs Python 3.11+") - return request.param - - @pytest.fixture def fresh_process_loop_state(): """ @@ -1180,7 +1165,6 @@ def _deprecation_warnings(caught: list[warnings.WarningMessage]) -> list[str]: return [str(w.message) for w in caught if issubclass(w.category, DeprecationWarning)] -@pytest.mark.usefixtures("event_loop_runner") class TestBaseAsyncOperator: class AsyncOperator(BaseAsyncOperator): async def aexecute(self, context): @@ -1219,9 +1203,8 @@ class TestBaseAsyncOperator: op.execute({}) -@pytest.mark.usefixtures("event_loop_runner") class TestEventLoop: - """``event_loop()`` hands synchronous code a loop it owns, through ``asyncio.Runner`` where Python has it.""" + """``event_loop()`` hands synchronous code a loop it owns, through ``asyncio.Runner``.""" @pytest.mark.usefixtures("fresh_process_loop_state") def test_creates_and_closes_own_loop_without_deprecation_warning(self):