|
1 | 1 | # SPDX-FileCopyrightText: Magenta ApS <https://magenta.dk> |
2 | 2 | # SPDX-License-Identifier: MPL-2.0 |
3 | 3 | import asyncio |
4 | | -import contextvars |
5 | 4 | import os |
6 | 5 | import secrets |
7 | | -import traceback |
8 | | -from asyncio import AbstractEventLoopPolicy |
9 | | -from asyncio import DefaultEventLoopPolicy |
10 | | -from asyncio import Task |
11 | 6 | from collections.abc import AsyncGenerator |
12 | 7 | from collections.abc import AsyncIterator |
13 | 8 | from collections.abc import Awaitable |
14 | 9 | from collections.abc import Callable |
15 | 10 | from collections.abc import Generator |
16 | 11 | from contextlib import asynccontextmanager |
17 | 12 | from dataclasses import dataclass |
18 | | -from functools import partial |
19 | 13 | from operator import itemgetter |
20 | | -from pathlib import Path |
21 | 14 | from typing import Any |
22 | 15 | from typing import AsyncContextManager |
23 | 16 | from typing import Protocol |
@@ -494,51 +487,6 @@ async def fixture_db( |
494 | 487 | yield session |
495 | 488 |
|
496 | 489 |
|
497 | | -@pytest.fixture(scope="session", autouse=True) |
498 | | -def event_loop_policy() -> AbstractEventLoopPolicy: |
499 | | - """Custom implementation of pytest-asyncio's event_loop_policy fixture[1]. |
500 | | -
|
501 | | - This fixture is used by pytest-asyncio to run test's setup/run/teardown. It |
502 | | - is needed to share contextvars between these stages; without it, |
503 | | - contextvars from async coroutine fixtures are not passed correctly to the |
504 | | - individual tests. See the issue[2] with solution implementation[3]. |
505 | | -
|
506 | | - The fixture name shadows the default fixture from pytest-asyncio, and thus |
507 | | - overrides it. Note that the links below reference overwriting the event_loop |
508 | | - fixture instead of the event_loop_policy -- this has been deprecated. |
509 | | -
|
510 | | - [1] https://github.com/pytest-dev/pytest-asyncio/blob/e92efad68146469228b3ac3478b254b692c6bc90/pytest_asyncio/plugin.py#L957-L970 |
511 | | - [2] https://github.com/pytest-dev/pytest-asyncio/issues/127 |
512 | | - [3] https://github.com/Donate4Fun/donate4fun/blob/cdf047365b7d2df83a952f5bb9544c29051fbdbd/tests/fixtures.py#L87-L113 |
513 | | - """ |
514 | | - |
515 | | - def task_factory(loop, coro, context=None): |
516 | | - # The task_factory breaks context isolation for asyncio tasks, so we need to |
517 | | - # check the calling context. |
518 | | - stack = traceback.extract_stack() |
519 | | - for frame in stack[-2::-1]: |
520 | | - package_name = Path(frame.filename).parts[-2] |
521 | | - if package_name != "asyncio": |
522 | | - if package_name == "pytest_asyncio": |
523 | | - # This function was called from pytest_asyncio, use shared context |
524 | | - break |
525 | | - else: |
526 | | - # This function was called from somewhere else, create context copy |
527 | | - context = None |
528 | | - break |
529 | | - return Task(coro, loop=loop, context=context) |
530 | | - |
531 | | - context = contextvars.copy_context() |
532 | | - |
533 | | - class CustomEventLoopPolicy(DefaultEventLoopPolicy): |
534 | | - def new_event_loop(self): |
535 | | - loop = super().new_event_loop() |
536 | | - loop.set_task_factory(partial(task_factory, context=context)) |
537 | | - return loop |
538 | | - |
539 | | - return CustomEventLoopPolicy() |
540 | | - |
541 | | - |
542 | 490 | @dataclass |
543 | 491 | class GQLResponse: |
544 | 492 | data: dict | None |
|
0 commit comments