|
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