|
2 | 2 | import typing |
3 | 3 | import os |
4 | 4 | import signal |
| 5 | +import sys |
5 | 6 |
|
6 | 7 |
|
7 | 8 | @pytest.fixture(scope="session", autouse=True) |
8 | | -def _cocoindex_env_fixture() -> typing.Generator[None, None, None]: |
9 | | - """Shutdown the subprocess pool at exit.""" |
| 9 | +def _cocoindex_windows_env_fixture( |
| 10 | + request: pytest.FixtureRequest, |
| 11 | +) -> typing.Generator[None, None, None]: |
| 12 | + """Shutdown the subprocess pool at exit on Windows.""" |
10 | 13 |
|
11 | 14 | yield |
12 | 15 |
|
| 16 | + if not sys.platform.startswith("win"): |
| 17 | + return |
| 18 | + |
13 | 19 | try: |
14 | 20 | print("Shutdown the subprocess pool at exit in hook.") |
15 | 21 | import cocoindex.subprocess_exec |
16 | 22 |
|
17 | | - if os.name == "nt": |
18 | | - original_sigint_handler = signal.getsignal(signal.SIGINT) |
19 | | - try: |
20 | | - signal.signal(signal.SIGINT, signal.SIG_IGN) |
21 | | - cocoindex.subprocess_exec.shutdown_pool_at_exit() |
22 | | - finally: |
23 | | - try: |
24 | | - signal.signal(signal.SIGINT, original_sigint_handler) |
25 | | - except ValueError: # noqa: BLE001 |
26 | | - pass |
27 | | - else: |
| 23 | + original_sigint_handler = signal.getsignal(signal.SIGINT) |
| 24 | + try: |
| 25 | + signal.signal(signal.SIGINT, signal.SIG_IGN) |
28 | 26 | cocoindex.subprocess_exec.shutdown_pool_at_exit() |
| 27 | + finally: |
| 28 | + try: |
| 29 | + signal.signal(signal.SIGINT, original_sigint_handler) |
| 30 | + except ValueError: # noqa: BLE001 |
| 31 | + pass |
| 32 | + |
29 | 33 | except (ImportError, AttributeError): # noqa: BLE001 |
30 | 34 | pass |
| 35 | + |
| 36 | + # If any test failed, let pytest exit normally with nonzero code |
| 37 | + if request.session.testsfailed == 0: |
| 38 | + os._exit(0) # immediate success exit (skips atexit/teardown) |
0 commit comments