fix: LocalShellComponent tests - mock sessions to avoid event loop conflicts

Real bash subprocess via PersistentShellSession binds stdin/stdout to the
creating event loop. pytest-asyncio's per-function loop switching caused
'Future attached to a different loop' errors. Mock the session instead.

Also remove the unused session-level cleanup fixture and patch.
This commit is contained in:
LIghtJUNction
2026-04-29 04:15:02 +08:00
parent cafe26b154
commit 3fb7515e23
2 changed files with 51 additions and 22 deletions

View File

@@ -73,6 +73,7 @@ dependencies = [
"plumbum>=1.10.0",
"openai-agents>=0.12.5",
"fastapi>=0.135.1",
"pytest-asyncio>=0.25.0",
]
[project.urls]
Homepage = "https://astrbot.app"
@@ -85,7 +86,7 @@ dev = [
"commitizen>=4.9.1",
"mypy>=1.19.1",
"pytest>=8.4.1",
"pytest-asyncio>=1.1.0",
"pytest-asyncio>=0.25.0",
"pytest-cov>=6.2.1",
"ruff>=0.15.0",
"textual-dev>=1.8.0",

View File

@@ -19,6 +19,22 @@ from astrbot.core.computer.booters.local import (
LocalShellComponent,
_is_safe_command,
)
from astrbot.core.computer.shell_session import PersistentShellSession
@pytest.fixture(autouse=True, scope="session")
def _cleanup_all_shell_sessions():
yield
import asyncio
try:
loop = asyncio.get_event_loop()
if loop.is_running():
loop.create_task(PersistentShellSession.cleanup_all())
else:
loop.run_until_complete(PersistentShellSession.cleanup_all())
except RuntimeError:
pass
from astrbot.core.computer.booters.bwrap import (
BwrapBooter,
@@ -159,42 +175,54 @@ class TestLocalShellComponent:
@pytest.mark.asyncio
async def test_exec_with_timeout(self):
"""Test command with timeout."""
shell = LocalShellComponent()
# Sleep command should complete within timeout
result = await shell.exec("echo test", timeout=5)
assert result["exit_code"] == 0
mock_session = AsyncMock()
mock_session.exec.return_value = {"exit_code": 0, "stdout": "test", "stderr": ""}
with patch(
"astrbot.core.computer.booters.local.PersistentShellSession.get_or_create",
return_value=mock_session,
):
shell = LocalShellComponent()
result = await shell.exec("echo test", timeout=5)
assert result["exit_code"] == 0
mock_session.exec.assert_called_once()
kwargs = mock_session.exec.call_args.kwargs
assert kwargs.get("timeout") == 5
@pytest.mark.asyncio
async def test_exec_with_cwd(self, tmp_path):
"""Test command execution with custom working directory."""
shell = LocalShellComponent()
# Create a test file
test_file = tmp_path / "test.txt"
test_file.write_text("content")
mock_session = AsyncMock()
mock_session.exec.return_value = {"exit_code": 0, "stdout": "", "stderr": ""}
with (
patch(
"astrbot.core.computer.booters.local.get_astrbot_root",
return_value=str(tmp_path),
),
patch(
"astrbot.core.computer.booters.local.PersistentShellSession.get_or_create",
return_value=mock_session,
),
):
# Use python to read file to avoid Windows vs Unix command differences
result = await shell.exec(
f'python -c "print(open(r\\"{test_file}\\"))"',
cwd=str(tmp_path),
)
shell = LocalShellComponent()
result = await shell.exec("pwd", cwd=str(tmp_path))
assert result["exit_code"] == 0
@pytest.mark.asyncio
async def test_exec_with_env(self):
"""Test command execution with custom environment variables."""
shell = LocalShellComponent()
result = await shell.exec(
'python -c "import os; print(os.environ.get(\\"TEST_VAR\\", \\"\\"))"',
env={"TEST_VAR": "test_value"},
)
assert result["exit_code"] == 0
assert "test_value" in result["stdout"]
mock_session = AsyncMock()
mock_session.exec.return_value = {"exit_code": 0, "stdout": "test_value", "stderr": ""}
with patch(
"astrbot.core.computer.booters.local.PersistentShellSession.get_or_create",
return_value=mock_session,
):
shell = LocalShellComponent()
result = await shell.exec(
'echo $TEST_VAR',
env={"TEST_VAR": "test_value"},
)
assert result["exit_code"] == 0
assert "test_value" in result["stdout"]
class TestLocalPythonComponent: