mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-01 01:10:21 +08:00
b5e29511ac
* fix: reliably kill shell process tree on Windows timeout Fixes #8809 * fix: remove redundant import and wrap taskkill in try/except - Remove 'import subprocess as _sp' (subprocess already imported at top) - Use subprocess.run directly with DEVNULL for stdout/stderr - Wrap taskkill in try/except to avoid masking original TimeoutExpired - If taskkill fails, cleanup failures don't prevent proc.wait() or re-raise https://buymeacoffee.com/muhamedfazalps * style: apply ruff formatting to local.py * test: fix shell component tests to match Popen-based implementation The tests were monkeypatching subprocess.run but the implementation now uses subprocess.Popen + communicate() for timeout handling. Updated tests to mock Popen instead. Fixes CI Unit Tests failure * fix: harden windows shell timeout cleanup --------- Co-authored-by: Soulter <905617992@qq.com>
137 lines
3.8 KiB
Python
137 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from astrbot.core.computer.booters import local as local_booter
|
|
from astrbot.core.computer.booters.local import LocalShellComponent
|
|
|
|
|
|
class _FakePopen:
|
|
def __init__(self, stdout: bytes, stderr: bytes = b"", returncode: int = 0):
|
|
self._stdout = stdout
|
|
self._stderr = stderr
|
|
self.returncode = returncode
|
|
self.pid = 12345
|
|
|
|
def communicate(self, timeout=None):
|
|
return self._stdout, self._stderr
|
|
|
|
def wait(self, timeout=None):
|
|
pass
|
|
|
|
|
|
class _FakeTaskkillResult:
|
|
def __init__(self, returncode: int):
|
|
self.returncode = returncode
|
|
|
|
|
|
def test_local_shell_component_decodes_utf8_output(monkeypatch):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout="技能内容".encode())
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "技能内容"
|
|
assert result["stderr"] == ""
|
|
assert result["exit_code"] == 0
|
|
|
|
|
|
def test_local_shell_component_prefers_utf8_before_windows_locale(
|
|
monkeypatch,
|
|
):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout="技能内容".encode())
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
monkeypatch.setattr(local_booter.os, "name", "nt", raising=False)
|
|
monkeypatch.setattr(
|
|
local_booter.locale,
|
|
"getpreferredencoding",
|
|
lambda _do_setlocale=False: "cp936",
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "技能内容"
|
|
assert result["stderr"] == ""
|
|
assert result["exit_code"] == 0
|
|
|
|
|
|
def test_local_shell_component_falls_back_to_gbk_on_windows(monkeypatch):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout="微博热搜".encode("gbk"))
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
monkeypatch.setattr(local_booter.os, "name", "nt", raising=False)
|
|
monkeypatch.setattr(
|
|
local_booter.locale,
|
|
"getpreferredencoding",
|
|
lambda _do_setlocale=False: "cp1252",
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "微博热搜"
|
|
assert result["stderr"] == ""
|
|
assert result["exit_code"] == 0
|
|
|
|
|
|
def test_local_shell_component_falls_back_to_utf8_replace(monkeypatch):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout=b"\xffabc")
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
monkeypatch.setattr(local_booter.os, "name", "posix", raising=False)
|
|
monkeypatch.setattr(
|
|
local_booter.locale,
|
|
"getpreferredencoding",
|
|
lambda _do_setlocale=False: "utf-8",
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "\ufffdabc"
|
|
|
|
|
|
def test_local_shell_component_falls_back_when_windows_taskkill_fails(monkeypatch):
|
|
class TimeoutPopen:
|
|
pid = 12345
|
|
|
|
def __init__(self):
|
|
self.killed = False
|
|
self.wait_timeout = None
|
|
|
|
def communicate(self, timeout=None):
|
|
raise subprocess.TimeoutExpired(cmd="dummy", timeout=timeout)
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
|
|
def wait(self, timeout=None):
|
|
self.wait_timeout = timeout
|
|
|
|
proc = TimeoutPopen()
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", lambda *_args, **_kwargs: proc)
|
|
monkeypatch.setattr(
|
|
subprocess,
|
|
"run",
|
|
lambda *_args, **_kwargs: _FakeTaskkillResult(returncode=1),
|
|
)
|
|
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
|
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
asyncio.run(LocalShellComponent().exec("dummy", timeout=1))
|
|
|
|
assert proc.killed
|
|
assert proc.wait_timeout == 5
|