FilesExpand file tree

main

/

conftest.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

49 lines (40 loc) · 1.81 KB

main

/

conftest.py

Copy path

Top

File metadata and controls

• Code
• Blame

49 lines (40 loc) · 1.81 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

# Copyright (c) 2026 dexpace and Omar Aljarrah.

# Licensed under the MIT License. See LICENSE.md in the repository root for details.

"""Workspace-root conftest for the dexpace SDK test suite.

Sole purpose right now: own and close an event loop on behalf of

``pytest-asyncio``'s ``_temporary_event_loop_policy`` fixture (plugin.py:618),

which calls ``asyncio.get_event_loop()`` to capture the "old" loop, gets a

freshly-created one back when no loop is current, and never closes it. The

leftover loop (and the socket pair backing its self-pipe) escalates as a

``PytestUnraisableExceptionWarning`` under our ``filterwarnings = ["error"]``

gate.

Pre-creating and registering an event loop here means that capture call

finds *our* loop instead of conjuring a new one, and the session finalizer

closes it deterministically.

Tracked upstream in pytest-asyncio; this conftest can go away once a fix

ships.

"""

from __future__ import annotations

import asyncio

from collections.abc import Iterator

import pytest

@pytest.fixture(scope="session", autouse=True)

def _own_default_event_loop() -> Iterator[None]:

loop = asyncio.new_event_loop()

asyncio.set_event_loop(loop)

try:

yield

finally:

# ``pytest_asyncio`` may have replaced the current loop; close

# whichever loop is current at teardown. ``asyncio.get_event_loop``

# (not the policy API — deprecated in 3.14) returns the set loop

# without warning on every supported version, and raises

# ``RuntimeError`` on 3.14+ when none is set.

try:

current = asyncio.get_event_loop()

except RuntimeError:

current = None

for candidate in {loop, current}:

if candidate is not None and not candidate.is_closed():

candidate.close()

asyncio.set_event_loop(None)