Sync from GitHub main #1
@@ -11,6 +11,8 @@ from pathlib import Path
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
from admin_server import send_admin_command
|
||||||
|
|
||||||
pytestmark = pytest.mark.skipif(os.name == "nt", reason="SIGTERM integration is POSIX-only")
|
pytestmark = pytest.mark.skipif(os.name == "nt", reason="SIGTERM integration is POSIX-only")
|
||||||
|
|
||||||
SERVER_DIR = Path(__file__).resolve().parents[1]
|
SERVER_DIR = Path(__file__).resolve().parents[1]
|
||||||
@@ -78,11 +80,16 @@ def test_sigterm_exits_cleanly(tmp_path: Path) -> None:
|
|||||||
time.sleep(0.1)
|
time.sleep(0.1)
|
||||||
assert connected, "server did not accept a test client in time"
|
assert connected, "server did not accept a test client in time"
|
||||||
|
|
||||||
# Admin probe over localhost only.
|
admin_deadline = time.time() + 5.0
|
||||||
with socket.create_connection(("127.0.0.1", admin_port), timeout=2.0) as admin:
|
response: dict[str, object] | None = None
|
||||||
admin.sendall(b'{"cmd":"ping"}\n')
|
while time.time() < admin_deadline:
|
||||||
response = admin.recv(4096)
|
try:
|
||||||
assert b'"ok":true' in response.replace(b" ", b"")
|
response = send_admin_command({"cmd": "ping"}, port=admin_port, timeout_seconds=0.5)
|
||||||
|
break
|
||||||
|
except (OSError, RuntimeError, ConnectionError):
|
||||||
|
time.sleep(0.05)
|
||||||
|
assert response is not None
|
||||||
|
assert response.get("ok") is True
|
||||||
|
|
||||||
process.send_signal(signal.SIGTERM)
|
process.send_signal(signal.SIGTERM)
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user