|
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.""" |
| 13 | + |
| 14 | + print("Platform: ", sys.platform) |
10 | 15 |
|
11 | 16 | yield |
12 | 17 |
|
| 18 | + if not sys.platform.startswith("win"): |
| 19 | + return |
| 20 | + |
13 | 21 | try: |
14 | 22 | print("Shutdown the subprocess pool at exit in hook.") |
15 | 23 | import cocoindex.subprocess_exec |
16 | 24 |
|
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: |
| 25 | + original_sigint_handler = signal.getsignal(signal.SIGINT) |
| 26 | + try: |
| 27 | + signal.signal(signal.SIGINT, signal.SIG_IGN) |
28 | 28 | cocoindex.subprocess_exec.shutdown_pool_at_exit() |
| 29 | + |
| 30 | + # If any test failed, let pytest exit normally with nonzero code |
| 31 | + if request.session.testsfailed == 0: |
| 32 | + os._exit(0) # immediate success exit (skips atexit/teardown) |
| 33 | + |
| 34 | + finally: |
| 35 | + try: |
| 36 | + signal.signal(signal.SIGINT, original_sigint_handler) |
| 37 | + except ValueError: # noqa: BLE001 |
| 38 | + pass |
| 39 | + |
29 | 40 | except (ImportError, AttributeError): # noqa: BLE001 |
30 | 41 | pass |
0 commit comments