102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from app.application.agent.tools import to_ai_provider_tools
|
|
from app.application.skill_executor import SkillExecutor
|
|
from app.application.tool_runtime.view import RunScopedToolView
|
|
from app.modules.skills.manifest import SkillManifest
|
|
from app.modules.tool_runtime.definition import ToolResult, ToolSpec
|
|
|
|
|
|
def _spec(name: str) -> ToolSpec:
|
|
return ToolSpec(
|
|
name=name,
|
|
description=f"{name} description",
|
|
risk_level="L1",
|
|
permission_policy=MagicMock(),
|
|
input_schema={
|
|
"type": "object",
|
|
"properties": {"value": {"type": "string"}},
|
|
"required": ["value"],
|
|
},
|
|
)
|
|
|
|
|
|
async def test_aliases_are_isolated_and_execute_through_base_runtime() -> None:
|
|
runtime = MagicMock()
|
|
runtime.list_specs.return_value = [_spec("shared_tool")]
|
|
runtime.execute = AsyncMock(
|
|
return_value=ToolResult(True, 0, "ok", "", 1, False, False, "trace_1")
|
|
)
|
|
first = RunScopedToolView(runtime)
|
|
second = RunScopedToolView(runtime)
|
|
|
|
assert first.bind_alias("skill:甲/shared_tool", "shared_tool", description_prefix="[甲] ")
|
|
assert second.bind_alias("skill:乙/shared_tool", "shared_tool", description_prefix="[乙] ")
|
|
|
|
first_names = {spec.name for spec in first.list_specs()}
|
|
second_names = {spec.name for spec in second.list_specs()}
|
|
result = await first.execute(
|
|
"skill:甲/shared_tool",
|
|
{"value": "x"},
|
|
run_id="run_1",
|
|
character_id="char_1",
|
|
)
|
|
|
|
assert "skill:甲/shared_tool" in first_names
|
|
assert "skill:乙/shared_tool" not in first_names
|
|
assert "skill:乙/shared_tool" in second_names
|
|
assert "skill:甲/shared_tool" not in second_names
|
|
assert result.stdout == "ok"
|
|
runtime.execute.assert_awaited_once_with(
|
|
"shared_tool",
|
|
{"value": "x"},
|
|
run_id="run_1",
|
|
character_id="char_1",
|
|
trace_id=None,
|
|
confirmed=False,
|
|
context=None,
|
|
)
|
|
runtime.register.assert_not_called()
|
|
runtime.unregister.assert_not_called()
|
|
|
|
|
|
def test_alias_keeps_target_input_schema() -> None:
|
|
runtime = MagicMock()
|
|
runtime.list_specs.return_value = [_spec("custom_tool")]
|
|
view = RunScopedToolView(runtime)
|
|
view.bind_alias("skill:研究/custom_tool", "custom_tool")
|
|
|
|
tools = to_ai_provider_tools(view.list_specs())
|
|
alias = next(tool for tool in tools if tool["function"]["name"] == "skill:研究/custom_tool")
|
|
|
|
assert alias["function"]["parameters"]["required"] == ["value"]
|
|
|
|
|
|
async def test_skill_executor_binds_only_current_run_aliases(
|
|
monkeypatch,
|
|
) -> None:
|
|
manifest = SkillManifest(
|
|
name="研究",
|
|
description="",
|
|
triggers=("研究",),
|
|
prompt="研究说明",
|
|
allowed_tools=("shared_tool", "missing"),
|
|
risk_level="L1",
|
|
)
|
|
|
|
async def load(*_args, **_kwargs):
|
|
return [manifest]
|
|
|
|
monkeypatch.setattr("app.application.skill_executor.load_skill_manifests", load)
|
|
runtime = MagicMock()
|
|
runtime.list_specs.return_value = [_spec("shared_tool")]
|
|
view = RunScopedToolView(runtime)
|
|
messages: list[str] = []
|
|
|
|
await SkillExecutor(view).match_and_activate("请研究", messages, run_id="run_1")
|
|
|
|
assert messages == ["【研究·能力注入】\n研究说明"]
|
|
assert "skill:研究/shared_tool" in view.specs_by_name()
|
|
assert "skill:研究/missing" not in view.specs_by_name()
|
|
runtime.register.assert_not_called()
|