mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-31 08:13:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

test(gateway): isolate lifecycle lock behavior

Browse Source
This commit is contained in:
Xubin Ren
2026-08-17 20:56:10 +08:00
parent 78aa29ffc9
commit f5125f3c07
2 changed files with 9 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+4 -5
View File
@@ -96,9 +96,9 @@ jobs:
os: windows-latest os: windows-latest
python-version: "3.14" python-version: "3.14"
coverage: false coverage: false
# Process lifecycle tests run serially below. Keep them out of # Real PowerShell/process-tree tests run serially below. Keep
# xdist so workers never share Windows console or process state. # them out of xdist so workers never share a Windows console.
pytest_args: "-n 2 --dist loadfile --ignore=tests/gateway/test_runtime.py --ignore=tests/tools/test_exec_platform.py" pytest_args: "-n 2 --dist loadfile --ignore=tests/tools/test_exec_platform.py"
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v4
@@ -148,11 +148,10 @@ jobs:
${{ matrix.pytest_args }} ${{ matrix.pytest_args }}
--durations=25 --durations-min=1.0 --durations=25 --durations-min=1.0
- name: Run Windows lifecycle compatibility tests - name: Run Windows process compatibility tests
if: runner.os == 'Windows' if: runner.os == 'Windows'
run: >- run: >-
uv run --no-sync python -m pytest uv run --no-sync python -m pytest
tests/gateway/test_runtime.py
tests/tools/test_exec_platform.py tests/tools/test_exec_platform.py
--durations=25 --durations-min=1.0 --durations=25 --durations-min=1.0
tests/gateway/test_runtime.py
+5
View File
@@ -206,6 +206,7 @@ def test_stop_reaps_an_owned_child_without_consuming_the_shutdown_timeout(
def test_concurrent_background_starts_create_only_one_process(tmp_path, monkeypatch): def test_concurrent_background_starts_create_only_one_process(tmp_path, monkeypatch):
first_spawned = threading.Event() first_spawned = threading.Event()
release_first = threading.Event() release_first = threading.Event()
lifecycle_lock = threading.Lock()
calls: list[list[str]] = [] calls: list[list[str]] = []
def fake_popen(command, **_kwargs): def fake_popen(command, **_kwargs):
@@ -231,6 +232,10 @@ def test_concurrent_background_starts_create_only_one_process(tmp_path, monkeypa
for runtime in (first, second): for runtime in (first, second):
monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True) monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
monkeypatch.setattr(runtime, "_process_identity", lambda _pid: 12345) monkeypatch.setattr(runtime, "_process_identity", lambda _pid: 12345)
# Exercise the runtime's critical section without mixing an OS-level
# file lock into a same-process thread scheduling test. FileLock is
# independently responsible for cross-process portability.
monkeypatch.setattr(runtime, "_lifecycle_lock", lambda: lifecycle_lock)
results = [] results = []
first_thread = threading.Thread( first_thread = threading.Thread(