File size: 11,144 Bytes
8ede856 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 | """Tests for profile-aware sandbox selection and conditional tool registration."""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import patch
import pytest
# ═══════════════════════════════════════════════════════════════
# ShipyardNeoBooter.capabilities
# ═══════════════════════════════════════════════════════════════
class TestShipyardNeoBooterCapabilities:
"""Test capabilities property on ShipyardNeoBooter."""
def _make_booter(self, sandbox_caps: list[str] | None = None):
from astrbot.core.computer.booters.shipyard_neo import ShipyardNeoBooter
booter = ShipyardNeoBooter(
endpoint_url="http://localhost:8114",
access_token="sk-bay-test",
)
if sandbox_caps is not None:
booter._sandbox = SimpleNamespace(capabilities=sandbox_caps)
return booter
def test_none_before_boot(self):
booter = self._make_booter()
assert booter.capabilities is None
def test_returns_tuple_after_boot(self):
booter = self._make_booter(["python", "shell", "filesystem"])
assert booter.capabilities == ("python", "shell", "filesystem")
assert isinstance(booter.capabilities, tuple)
def test_includes_browser_when_present(self):
booter = self._make_booter(["python", "shell", "filesystem", "browser"])
assert "browser" in booter.capabilities
def test_no_browser_when_absent(self):
booter = self._make_booter(["python", "shell", "filesystem"])
assert "browser" not in booter.capabilities
def test_returns_immutable(self):
"""Verify capabilities returns an immutable tuple."""
booter = self._make_booter(["python"])
caps = booter.capabilities
assert isinstance(caps, tuple)
with pytest.raises(AttributeError):
caps.append("mutated") # type: ignore[attr-defined]
# ═══════════════════════════════════════════════════════════════
# _apply_sandbox_tools — conditional browser tool registration
# ═══════════════════════════════════════════════════════════════
def _make_config(booter_type: str = "shipyard_neo"):
return SimpleNamespace(
sandbox_cfg={"booter": booter_type},
)
def _make_req():
return SimpleNamespace(func_tool=None, system_prompt="")
def _import_apply_sandbox_tools():
"""Import _apply_sandbox_tools, skipping if circular-import fails."""
try:
from astrbot.core.astr_main_agent import _apply_sandbox_tools
return _apply_sandbox_tools
except ImportError:
pytest.skip("Cannot import _apply_sandbox_tools (circular import in test env)")
class TestApplySandboxToolsConditional:
"""Verify browser tools are conditionally registered."""
def _tool_names(self, req) -> set[str]:
"""Extract tool names from a request's func_tool."""
if req.func_tool is None:
return set()
return {t.name for t in req.func_tool.tools}
def test_no_session_registers_all(self):
"""First request (no booted session) → all tools including browser."""
fn = _import_apply_sandbox_tools()
config = _make_config("shipyard_neo")
req = _make_req()
with patch(
"astrbot.core.computer.computer_client.session_booter", {}
):
fn(config, req, "session-1")
names = self._tool_names(req)
assert "astrbot_execute_browser" in names
assert "astrbot_execute_browser_batch" in names
assert "astrbot_run_browser_skill" in names
def test_with_browser_capability(self):
"""Booted session with browser capability → browser tools registered."""
fn = _import_apply_sandbox_tools()
config = _make_config("shipyard_neo")
req = _make_req()
fake_booter = SimpleNamespace(
capabilities=["python", "shell", "filesystem", "browser"]
)
with patch(
"astrbot.core.computer.computer_client.session_booter",
{"session-1": fake_booter},
):
fn(config, req, "session-1")
names = self._tool_names(req)
assert "astrbot_execute_browser" in names
def test_without_browser_capability(self):
"""Booted session WITHOUT browser capability → browser tools NOT registered."""
fn = _import_apply_sandbox_tools()
config = _make_config("shipyard_neo")
req = _make_req()
fake_booter = SimpleNamespace(
capabilities=["python", "shell", "filesystem"]
)
with patch(
"astrbot.core.computer.computer_client.session_booter",
{"session-1": fake_booter},
):
fn(config, req, "session-1")
names = self._tool_names(req)
assert "astrbot_execute_browser" not in names
assert "astrbot_execute_browser_batch" not in names
assert "astrbot_run_browser_skill" not in names
# Skill tools should still be registered
assert "astrbot_get_execution_history" in names
def test_skill_tools_always_registered(self):
"""Skill lifecycle tools are registered regardless of capabilities."""
fn = _import_apply_sandbox_tools()
config = _make_config("shipyard_neo")
req = _make_req()
fake_booter = SimpleNamespace(capabilities=["python"])
with patch(
"astrbot.core.computer.computer_client.session_booter",
{"session-1": fake_booter},
):
fn(config, req, "session-1")
names = self._tool_names(req)
assert "astrbot_create_skill_candidate" in names
assert "astrbot_promote_skill_candidate" in names
# ═══════════════════════════════════════════════════════════════
# _resolve_profile
# ═══════════════════════════════════════════════════════════════
class TestResolveProfile:
"""Test smart profile selection logic."""
def _make_booter(self, profile: str = "python-default"):
from astrbot.core.computer.booters.shipyard_neo import ShipyardNeoBooter
return ShipyardNeoBooter(
endpoint_url="http://localhost:8114",
access_token="sk-bay-test",
profile=profile,
)
@pytest.mark.asyncio
async def test_user_specified_profile_honoured(self):
"""User explicitly sets a non-default profile → use it directly."""
booter = self._make_booter(profile="browser-python")
client = SimpleNamespace() # list_profiles should NOT be called
result = await booter._resolve_profile(client)
assert result == "browser-python"
@pytest.mark.asyncio
async def test_selects_browser_profile(self):
"""When multiple profiles available, prefer one with browser."""
async def _mock_list_profiles():
return SimpleNamespace(
items=[
SimpleNamespace(
id="python-default",
capabilities=["python", "shell", "filesystem"],
),
SimpleNamespace(
id="browser-python",
capabilities=["python", "shell", "filesystem", "browser"],
),
]
)
booter = self._make_booter()
client = SimpleNamespace(list_profiles=_mock_list_profiles)
result = await booter._resolve_profile(client)
assert result == "browser-python"
@pytest.mark.asyncio
async def test_falls_back_to_default_on_api_error(self):
"""API error → graceful fallback to python-default."""
async def _failing_list_profiles():
raise ConnectionError("Bay unreachable")
booter = self._make_booter()
client = SimpleNamespace(list_profiles=_failing_list_profiles)
result = await booter._resolve_profile(client)
assert result == "python-default"
@pytest.mark.asyncio
async def test_falls_back_on_empty_profiles(self):
"""Empty profile list → python-default."""
async def _empty_list_profiles():
return SimpleNamespace(items=[])
booter = self._make_booter()
client = SimpleNamespace(list_profiles=_empty_list_profiles)
result = await booter._resolve_profile(client)
assert result == "python-default"
@pytest.mark.asyncio
async def test_single_profile_selected(self):
"""Only one profile available → use it."""
async def _single_profile():
return SimpleNamespace(
items=[
SimpleNamespace(
id="python-data",
capabilities=["python", "shell", "filesystem"],
),
]
)
booter = self._make_booter()
client = SimpleNamespace(list_profiles=_single_profile)
result = await booter._resolve_profile(client)
assert result == "python-data"
@pytest.mark.asyncio
async def test_auth_error_not_silenced(self):
"""UnauthorizedError must propagate, not be downgraded to fallback."""
from shipyard_neo.errors import UnauthorizedError
async def _unauthorized_list_profiles():
raise UnauthorizedError("bad token")
booter = self._make_booter()
client = SimpleNamespace(list_profiles=_unauthorized_list_profiles)
with pytest.raises(UnauthorizedError):
await booter._resolve_profile(client)
# ═══════════════════════════════════════════════════════════════
# ComputerBooter base class
# ═══════════════════════════════════════════════════════════════
class TestBaseComputerBooter:
"""Verify base class defaults."""
def test_capabilities_default_none(self):
from astrbot.core.computer.booters.base import ComputerBooter
booter = ComputerBooter()
assert booter.capabilities is None
def test_browser_default_none(self):
from astrbot.core.computer.booters.base import ComputerBooter
booter = ComputerBooter()
assert booter.browser is None
|