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): track the claimed process identity

Browse Source
This commit is contained in:
Xubin Ren
2026-08-17 20:56:10 +08:00
parent 6955d709d6
commit 7838e244ba
1 changed files with 7 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/gateway/test_runtime.py
+7 -7
View File
@@ -67,7 +67,7 @@ try:
if os.name != "nt": if os.name != "nt":
signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGTERM, stop)
with runtime.foreground_instance(GatewayStartOptions(port=18790)): with runtime.foreground_instance(GatewayStartOptions(port=18790)):
marker.write_text("claimed", encoding="utf-8") marker.write_text(f"claimed:{os.getpid()}", encoding="utf-8")
time.sleep(duration) time.sleep(duration)
except GatewayAlreadyRunningError: except GatewayAlreadyRunningError:
marker.write_text("occupied", encoding="utf-8") marker.write_text("occupied", encoding="utf-8")
@@ -105,13 +105,13 @@ def _foreground_child(
def _wait_for_claim( def _wait_for_claim(
process: subprocess.Popen[str], process: subprocess.Popen[str],
marker: Path, marker: Path,
) -> None: ) -> int:
deadline = time.monotonic() + 3 deadline = time.monotonic() + 3
while time.monotonic() < deadline: while time.monotonic() < deadline:
if marker.exists(): if marker.exists():
detail = marker.read_text(encoding="utf-8") detail = marker.read_text(encoding="utf-8")
if detail == "claimed": if detail.startswith("claimed:"):
return return int(detail.partition(":")[2])
pytest.fail(f"gateway claim failed: returncode={process.poll()}, {detail}") pytest.fail(f"gateway claim failed: returncode={process.poll()}, {detail}")
if process.poll() is not None: if process.poll() is not None:
break break
@@ -305,7 +305,7 @@ def test_competing_foreground_claim_preserves_the_live_gateway(tmp_path):
first_marker = tmp_path / "first.marker" first_marker = tmp_path / "first.marker"
first = _foreground_child(tmp_path, 30, first_marker) first = _foreground_child(tmp_path, 30, first_marker)
try: try:
_wait_for_claim(first, first_marker) first_pid = _wait_for_claim(first, first_marker)
second_marker = tmp_path / "second.marker" second_marker = tmp_path / "second.marker"
second = _foreground_child(tmp_path, 0, second_marker) second = _foreground_child(tmp_path, 0, second_marker)
try: try:
@@ -318,8 +318,8 @@ def test_competing_foreground_claim_preserves_the_live_gateway(tmp_path):
second.wait(timeout=3) second.wait(timeout=3)
state = json.loads(runtime.paths.state_path.read_text(encoding="utf-8")) state = json.loads(runtime.paths.state_path.read_text(encoding="utf-8"))
assert state["pid"] == first.pid assert state["pid"] == first_pid
assert runtime.status().pid == first.pid assert runtime.status().pid == first_pid
finally: finally:
if first.poll() is None: if first.poll() is None:
first.terminate() first.terminate()