ik/rf4-spotter
1
0
Fork 0
Code Issues Pull Requests Actions 7 Packages Projects Releases Wiki Activity

test: clean up concurrent cooldown processes

Browse Source
This commit is contained in:
ik
2026-09-11 07:56:01 +07:00
parent 8f76f70409
commit 51728bf3f2
1 changed files with 21 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_community_cli.py
+7 -3
View File
@@ -137,9 +137,9 @@ def test_check_and_reserve_atomic_under_concurrent_access(tmp_path: Path) -> Non
import multiprocessing import multiprocessing
state_file = tmp_path / "concurrent.json" state_file = tmp_path / "concurrent.json"
results = multiprocessing.Manager().list() # Explicitly shut down the manager and never leak a timed-out child.
with multiprocessing.Manager() as manager:
# Launch 3 processes simultaneously results = manager.list()
processes = [] processes = []
for i in range(3): for i in range(3):
args = (i, str(state_file), results) args = (i, str(state_file), results)
@@ -149,6 +149,10 @@ def test_check_and_reserve_atomic_under_concurrent_access(tmp_path: Path) -> Non
p.start() p.start()
for p in processes: for p in processes:
p.join(timeout=10) p.join(timeout=10)
for p in processes:
if p.is_alive():
p.terminate()
p.join(timeout=2)
# At most one should succeed # At most one should succeed
ok_count = sum(1 for _, r in results if r == "ok") ok_count = sum(1 for _, r in results if r == "ok")