|
25 | 25 | from typing import Any |
26 | 26 | from unittest.mock import patch |
27 | 27 |
|
| 28 | +import pytest |
| 29 | + |
28 | 30 | from apache_beam.utils import multi_process_shared |
29 | 31 |
|
30 | 32 |
|
| 33 | +@pytest.fixture(autouse=True) |
| 34 | +def isolate_multi_process_shared_tests(tmp_path, monkeypatch): |
| 35 | + """Isolates MultiProcessShared tests by using a unique temp directory per test. |
| 36 | +
|
| 37 | + This prevents tests running in parallel (e.g. with pytest-xdist) from |
| 38 | + interfering with each other by writing to the same shared default temp directory. |
| 39 | + """ |
| 40 | + orig_init = multi_process_shared.MultiProcessShared.__init__ |
| 41 | + |
| 42 | + def mock_init(self, constructor, tag, *args, **kwargs): |
| 43 | + if 'path' not in kwargs: |
| 44 | + kwargs['path'] = str(tmp_path) |
| 45 | + return orig_init(self, constructor, tag, *args, **kwargs) |
| 46 | + |
| 47 | + monkeypatch.setattr( |
| 48 | + multi_process_shared.MultiProcessShared, '__init__', mock_init) |
| 49 | + |
| 50 | + |
31 | 51 | class CallableCounter(object): |
32 | 52 | def __init__(self, start=0): |
33 | 53 | self.running = start |
@@ -285,23 +305,6 @@ def test_release_always_proxy(self): |
285 | 305 |
|
286 | 306 |
|
287 | 307 | class MultiProcessSharedSpawnProcessTest(unittest.TestCase): |
288 | | - def setUp(self): |
289 | | - tempdir = tempfile.gettempdir() |
290 | | - for tag in ['basic', |
291 | | - 'main', |
292 | | - 'to_delete', |
293 | | - 'to_keep', |
294 | | - 'mix1', |
295 | | - 'mix2', |
296 | | - 'test_process_exit', |
297 | | - 'thundering_herd_test', |
298 | | - 'transient_test']: |
299 | | - for ext in ['', '.address', '.address.error']: |
300 | | - try: |
301 | | - os.remove(os.path.join(tempdir, tag + ext)) |
302 | | - except OSError: |
303 | | - pass |
304 | | - |
305 | 308 | def tearDown(self): |
306 | 309 | for p in multiprocessing.active_children(): |
307 | 310 | if p.is_alive(): |
|
0 commit comments