wehub-resource-sync
303 lines
9.5 KiB
Python
303 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from yuxi.agents.skills import service as skill_service
|
|
from yuxi.agents.toolkits.buildin import install_skill as exported_install_skill
|
|
|
|
install_skill_module = importlib.import_module("yuxi.agents.toolkits.buildin.install_skill")
|
|
|
|
|
|
class _AsyncSessionContext:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def __aenter__(self):
|
|
return self.db
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
|
|
def _runtime(**context_values):
|
|
return SimpleNamespace(context=SimpleNamespace(**context_values))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_skill_from_sandbox_installs_as_current_user_private_skill(monkeypatch, tmp_path: Path):
|
|
assert exported_install_skill.name == "install_skill"
|
|
|
|
calls = {}
|
|
db = SimpleNamespace()
|
|
source_dir = tmp_path / "demo-skill"
|
|
|
|
def prepare_skill_from_sandbox(source, thread_id, uid, staging_root):
|
|
calls["prepare"] = {
|
|
"source": source,
|
|
"thread_id": thread_id,
|
|
"uid": uid,
|
|
"staging_root": staging_root,
|
|
}
|
|
return source_dir, "demo-skill"
|
|
|
|
async def import_skill_dir(db_arg, **kwargs):
|
|
calls["import"] = {"db": db_arg, **kwargs}
|
|
return SimpleNamespace(slug="demo-skill")
|
|
|
|
async def enable_skills(db_arg, thread_id, uid, skill_slugs):
|
|
calls["enable"] = {"db": db_arg, "thread_id": thread_id, "uid": uid, "skill_slugs": skill_slugs}
|
|
return True
|
|
|
|
def sync_thread_readable_skills(thread_id, skills):
|
|
calls["sync"] = {"thread_id": thread_id, "skills": skills}
|
|
|
|
monkeypatch.setattr(
|
|
install_skill_module,
|
|
"_prepare_skill_from_sandbox",
|
|
prepare_skill_from_sandbox,
|
|
)
|
|
monkeypatch.setattr(
|
|
install_skill_module,
|
|
"_enable_skills_in_current_config",
|
|
enable_skills,
|
|
)
|
|
monkeypatch.setattr(
|
|
install_skill_module.pg_manager,
|
|
"get_async_session_context",
|
|
lambda: _AsyncSessionContext(db),
|
|
)
|
|
monkeypatch.setattr(skill_service, "import_skill_dir", import_skill_dir)
|
|
monkeypatch.setattr(skill_service, "sync_thread_readable_skills", sync_thread_readable_skills)
|
|
|
|
result = await install_skill_module._run_install_task(
|
|
" /home/gem/user-data/workspace/demo-skill ",
|
|
_runtime(uid="normal-user", thread_id="thread-1", skills=["existing-skill"]),
|
|
"tool-1",
|
|
)
|
|
|
|
assert result.update["activated_skills"] == ["demo-skill"]
|
|
assert "成功安装并激活技能" in result.update["messages"][0].content
|
|
assert calls["prepare"]["uid"] == "normal-user"
|
|
assert calls["import"]["created_by"] == "normal-user"
|
|
assert calls["import"]["share_config"] == {
|
|
"access_level": "user",
|
|
"department_ids": [],
|
|
"user_uids": ["normal-user"],
|
|
}
|
|
assert calls["prepare"]["source"] == "/home/gem/user-data/workspace/demo-skill"
|
|
assert calls["enable"] == {
|
|
"db": db,
|
|
"thread_id": "thread-1",
|
|
"uid": "normal-user",
|
|
"skill_slugs": ["demo-skill"],
|
|
}
|
|
assert calls["sync"] == {"thread_id": "thread-1", "skills": ["existing-skill", "demo-skill"]}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_skill_rejects_subagent_runtime_before_install(monkeypatch):
|
|
def fail_get_session():
|
|
raise AssertionError("子智能体运行态不应访问数据库或执行安装")
|
|
|
|
monkeypatch.setattr(
|
|
install_skill_module.pg_manager,
|
|
"get_async_session_context",
|
|
fail_get_session,
|
|
)
|
|
|
|
result = await install_skill_module._run_install_task(
|
|
"/home/gem/user-data/workspace/demo-skill",
|
|
_runtime(uid="user-1", thread_id="child-thread", is_subagent_runtime=True),
|
|
"tool-1",
|
|
)
|
|
|
|
assert "只能在主智能体中使用" in result.update["messages"][0].content
|
|
assert "activated_skills" not in result.update
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_skill_git_source_requires_skill_names():
|
|
result = await install_skill_module._run_install_task(
|
|
"owner/repo",
|
|
_runtime(uid="user-1", thread_id="thread-1"),
|
|
"tool-1",
|
|
)
|
|
|
|
assert "必须通过 skill_names 指定技能名称" in result.update["messages"][0].content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_skill_rejects_empty_source():
|
|
result = await install_skill_module._run_install_task(
|
|
" ",
|
|
_runtime(uid="user-1", thread_id="thread-1"),
|
|
"tool-1",
|
|
)
|
|
|
|
assert "Skill 来源不能为空" in result.update["messages"][0].content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_skills_updates_current_user_owned_agent_config(monkeypatch):
|
|
conv = SimpleNamespace(uid="user-1", agent_id="agent-1")
|
|
agent = SimpleNamespace(
|
|
created_by="user-1",
|
|
config_json={"context": {"skills": ["existing-skill"], "model": "provider:model"}},
|
|
)
|
|
calls = {}
|
|
|
|
class FakeConversationRepository:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_conversation_by_thread_id(self, thread_id):
|
|
calls["thread_id"] = thread_id
|
|
return conv
|
|
|
|
class FakeAgentRepository:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_by_slug(self, slug):
|
|
calls["agent_slug"] = slug
|
|
return agent
|
|
|
|
async def update(self, agent_arg, **kwargs):
|
|
calls["update"] = {"agent": agent_arg, **kwargs}
|
|
return agent_arg
|
|
|
|
monkeypatch.setattr(install_skill_module, "ConversationRepository", FakeConversationRepository)
|
|
monkeypatch.setattr(install_skill_module, "AgentRepository", FakeAgentRepository)
|
|
|
|
result = await install_skill_module._enable_skills_in_current_config(
|
|
SimpleNamespace(),
|
|
"thread-1",
|
|
"user-1",
|
|
["existing-skill", "new-skill"],
|
|
)
|
|
|
|
assert result is True
|
|
assert calls["thread_id"] == "thread-1"
|
|
assert calls["agent_slug"] == "agent-1"
|
|
assert calls["update"]["updated_by"] == "user-1"
|
|
assert calls["update"]["config_json"] == {
|
|
"context": {"skills": ["existing-skill", "new-skill"], "model": "provider:model"}
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_skills_does_not_update_agent_not_owned_by_current_user(monkeypatch):
|
|
conv = SimpleNamespace(uid="user-1", agent_id="shared-agent")
|
|
agent = SimpleNamespace(created_by="admin", config_json={"context": {}})
|
|
calls = {}
|
|
|
|
class FakeConversationRepository:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_conversation_by_thread_id(self, _thread_id):
|
|
return conv
|
|
|
|
class FakeAgentRepository:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_by_slug(self, _slug):
|
|
return agent
|
|
|
|
async def update(self, *_args, **_kwargs):
|
|
calls["updated"] = True
|
|
|
|
monkeypatch.setattr(install_skill_module, "ConversationRepository", FakeConversationRepository)
|
|
monkeypatch.setattr(install_skill_module, "AgentRepository", FakeAgentRepository)
|
|
|
|
result = await install_skill_module._enable_skills_in_current_config(
|
|
SimpleNamespace(),
|
|
"thread-1",
|
|
"user-1",
|
|
["new-skill"],
|
|
)
|
|
|
|
assert result is False
|
|
assert "updated" not in calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_skills_does_not_update_mismatched_runtime_uid(monkeypatch):
|
|
conv = SimpleNamespace(uid="user-1", agent_id="agent-1")
|
|
calls = {}
|
|
|
|
class FakeConversationRepository:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_conversation_by_thread_id(self, _thread_id):
|
|
return conv
|
|
|
|
class FakeAgentRepository:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_by_slug(self, *_args):
|
|
calls["loaded_agent"] = True
|
|
|
|
monkeypatch.setattr(install_skill_module, "ConversationRepository", FakeConversationRepository)
|
|
monkeypatch.setattr(install_skill_module, "AgentRepository", FakeAgentRepository)
|
|
|
|
result = await install_skill_module._enable_skills_in_current_config(
|
|
SimpleNamespace(),
|
|
"thread-1",
|
|
"other-user",
|
|
["new-skill"],
|
|
)
|
|
|
|
assert result is False
|
|
assert "loaded_agent" not in calls
|
|
|
|
|
|
def test_prepare_skill_invalid_virtual_path_does_not_fallback_to_sandbox(monkeypatch, tmp_path: Path):
|
|
calls = {}
|
|
|
|
def resolve_virtual_path(*_args, **_kwargs):
|
|
raise ValueError("path traversal detected")
|
|
|
|
class FakeProvisionerSandboxBackend:
|
|
def __init__(self, *_args, **_kwargs):
|
|
calls["fallback"] = True
|
|
|
|
monkeypatch.setattr(
|
|
"yuxi.agents.backends.sandbox.resolve_virtual_path",
|
|
resolve_virtual_path,
|
|
)
|
|
monkeypatch.setattr(
|
|
"yuxi.agents.backends.sandbox.ProvisionerSandboxBackend",
|
|
FakeProvisionerSandboxBackend,
|
|
)
|
|
monkeypatch.setattr(skill_service, "is_valid_skill_slug", lambda _slug: True)
|
|
|
|
with pytest.raises(ValueError, match="path traversal detected"):
|
|
install_skill_module._prepare_skill_from_sandbox(
|
|
"/home/gem/user-data/workspace/demo-skill",
|
|
"thread-1",
|
|
"user-1",
|
|
tmp_path,
|
|
)
|
|
|
|
assert "fallback" not in calls
|
|
|
|
|
|
def test_collect_sandbox_file_paths_rejects_more_than_1000_files():
|
|
class FakeBackend:
|
|
def ls(self, _remote_dir):
|
|
return SimpleNamespace(
|
|
error=None,
|
|
entries=[{"path": f"/skill/file-{idx}.txt", "is_dir": False} for idx in range(1001)],
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="最多 1000 个文件"):
|
|
install_skill_module._collect_sandbox_file_paths(FakeBackend(), "/skill")
|