fix(test): 修复或删除一些因为重构报错的测试文件
✅ test_graph_unit.py - 修复了 config mock 和 driver/status 属性问题 ✅ test_skill_service.py - 修复了函数名变更和导入问题 ✅ test_skills_backend.py - 删除了过时的 create_agent_composite_backend 测试 ✅ 删除了 test_runtime_config_middleware_skills.py（API 完全重构）
This commit is contained in:
肖泽涛
parent
bb41056900
commit
ad41efef94
@ -6,7 +6,7 @@ import sys
|
||||
# Add project root to path
|
||||
sys.path.append(os.getcwd())
|
||||
|
||||
from src.knowledge.services.upload_graph_service import UploadGraphService
|
||||
from src.knowledge.graphs.upload_graph_service import UploadGraphService
|
||||
|
||||
# For backward compatibility with the existing test
|
||||
GraphDatabase = UploadGraphService
|
||||
@ -40,64 +40,68 @@ async def test_txt_add_vector_entity_parsing():
|
||||
mock_embed_model = MagicMock()
|
||||
mock_select_model.return_value = mock_embed_model
|
||||
|
||||
# Instantiate GraphDatabase with mocked driver
|
||||
# We also need to patch Neo4jConnectionManager in the init
|
||||
with patch("src.knowledge.services.upload_graph_service.Neo4jConnectionManager") as mock_connection_manager:
|
||||
# Mock the connection manager to return our mocked driver
|
||||
mock_connection_manager.return_value.driver = mock_driver
|
||||
mock_connection_manager.return_value.status = "open"
|
||||
# Create a mock connection object with driver and status as attributes
|
||||
# (not a ConnectionManager class, just a simple mock object)
|
||||
mock_connection = MagicMock()
|
||||
mock_connection.driver = mock_driver
|
||||
mock_connection.status = "open"
|
||||
|
||||
gd = GraphDatabase(mock_connection_manager.return_value)
|
||||
# Manually set properties just in case init didn't work as expected due to other mocks
|
||||
gd.driver = mock_driver
|
||||
gd.status = "open"
|
||||
gd.embed_model_name = "test_model" # avoid config check issues if possible
|
||||
# Instantiate GraphDatabase with mocked connection
|
||||
gd = GraphDatabase(mock_connection)
|
||||
# Set embed_model_name directly (this is a settable attribute)
|
||||
gd.embed_model_name = "test_model"
|
||||
|
||||
# Mock config to match
|
||||
with patch("src.config.embed_model", "test_model"):
|
||||
with patch("src.config.embed_model_names", {"test_model": MagicMock(dimension=1024)}):
|
||||
# Test data: Mixed format
|
||||
triples = [
|
||||
# Legacy format
|
||||
{"h": "A", "r": "KNOWS", "t": "B"},
|
||||
# Extended format
|
||||
{
|
||||
"h": {"name": "C", "age": 30},
|
||||
"r": {"type": "LIKES", "weight": 0.8},
|
||||
"t": {"name": "D", "role": "User"},
|
||||
},
|
||||
]
|
||||
# Mock config where it's imported in upload_graph_service.py
|
||||
# The import is `from src import config`, so patch at the usage location
|
||||
mock_config = MagicMock()
|
||||
mock_config.embed_model = "test_model"
|
||||
mock_embed_info = MagicMock()
|
||||
mock_embed_info.dimension = 1024
|
||||
mock_config.embed_model_names = {"test_model": mock_embed_info}
|
||||
|
||||
# Run the method
|
||||
await gd.txt_add_vector_entity(triples)
|
||||
with patch("src.knowledge.graphs.upload_graph_service.config", mock_config):
|
||||
# Test data: Mixed format
|
||||
triples = [
|
||||
# Legacy format
|
||||
{"h": "A", "r": "KNOWS", "t": "B"},
|
||||
# Extended format
|
||||
{
|
||||
"h": {"name": "C", "age": 30},
|
||||
"r": {"type": "LIKES", "weight": 0.8},
|
||||
"t": {"name": "D", "role": "User"},
|
||||
},
|
||||
]
|
||||
|
||||
# Verify calls to mock_tx.run
|
||||
merge_calls = []
|
||||
for call in mock_tx.run.call_args_list:
|
||||
args, kwargs = call
|
||||
query = args[0] if args else kwargs.get("query", "")
|
||||
if "MERGE (h:Entity:Upload" in query:
|
||||
# The args are passed as kwargs to run: h_name=..., etc.
|
||||
merge_calls.append(kwargs)
|
||||
# Run the method
|
||||
await gd.txt_add_vector_entity(triples)
|
||||
|
||||
assert len(merge_calls) == 2, f"Expected 2 merge calls, got {len(merge_calls)}"
|
||||
# Verify calls to mock_tx.run
|
||||
merge_calls = []
|
||||
for call in mock_tx.run.call_args_list:
|
||||
args, kwargs = call
|
||||
query = args[0] if args else kwargs.get("query", "")
|
||||
if "MERGE (h:Entity:Upload" in query:
|
||||
# The args are passed as kwargs to run: h_name=..., etc.
|
||||
merge_calls.append(kwargs)
|
||||
|
||||
# Call 1 (Legacy)
|
||||
call1 = merge_calls[0]
|
||||
assert call1["h_name"] == "A"
|
||||
assert call1["h_props"] == {}
|
||||
assert call1["t_name"] == "B"
|
||||
assert call1["t_props"] == {}
|
||||
assert call1["r_type"] == "KNOWS"
|
||||
assert call1["r_props"] == {}
|
||||
assert len(merge_calls) == 2, f"Expected 2 merge calls, got {len(merge_calls)}"
|
||||
|
||||
# Call 2 (Extended)
|
||||
call2 = merge_calls[1]
|
||||
assert call2["h_name"] == "C"
|
||||
assert call2["h_props"] == {"age": 30}
|
||||
assert call2["t_name"] == "D"
|
||||
assert call2["t_props"] == {"role": "User"}
|
||||
assert call2["r_type"] == "LIKES"
|
||||
assert call2["r_props"] == {"weight": 0.8}
|
||||
# Call 1 (Legacy)
|
||||
call1 = merge_calls[0]
|
||||
assert call1["h_name"] == "A"
|
||||
assert call1["h_props"] == {}
|
||||
assert call1["t_name"] == "B"
|
||||
assert call1["t_props"] == {}
|
||||
assert call1["r_type"] == "KNOWS"
|
||||
assert call1["r_props"] == {}
|
||||
|
||||
print("Verification passed!")
|
||||
# Call 2 (Extended)
|
||||
call2 = merge_calls[1]
|
||||
assert call2["h_name"] == "C"
|
||||
assert call2["h_props"] == {"age": 30}
|
||||
assert call2["t_name"] == "D"
|
||||
assert call2["t_props"] == {"role": "User"}
|
||||
assert call2["r_type"] == "LIKES"
|
||||
assert call2["r_props"] == {"weight": 0.8}
|
||||
|
||||
print("Verification passed!")
|
||||
@ -1,286 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import SystemMessage, ToolMessage
|
||||
from langgraph.types import Command
|
||||
|
||||
import src.agents.common.middlewares.runtime_config_middleware as runtime_middleware
|
||||
from src.agents.common.middlewares.runtime_config_middleware import RuntimeConfigMiddleware
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeTool:
|
||||
name: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeRequest:
|
||||
runtime: Any
|
||||
tools: list[Any]
|
||||
system_message: SystemMessage
|
||||
state: dict[str, Any]
|
||||
|
||||
def override(self, **kwargs):
|
||||
return _FakeRequest(
|
||||
runtime=kwargs.get("runtime", self.runtime),
|
||||
tools=kwargs.get("tools", self.tools),
|
||||
system_message=kwargs.get("system_message", self.system_message),
|
||||
state=kwargs.get("state", self.state),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeToolCallRequest:
|
||||
tool_call: dict[str, Any]
|
||||
runtime: Any
|
||||
state: dict[str, Any]
|
||||
|
||||
|
||||
async def _echo_handler(request):
|
||||
return request
|
||||
|
||||
|
||||
def _build_request(
|
||||
*,
|
||||
skills: list[str],
|
||||
tools: list[str],
|
||||
system_prompt: str = "你是助手",
|
||||
state: dict[str, Any] | None = None,
|
||||
skill_snapshot: dict[str, Any] | None = None,
|
||||
) -> _FakeRequest:
|
||||
context = SimpleNamespace(
|
||||
system_prompt=system_prompt,
|
||||
skills=skills,
|
||||
tools=[],
|
||||
knowledges=[],
|
||||
mcps=[],
|
||||
)
|
||||
if skill_snapshot is not None:
|
||||
context.skill_session_snapshot = skill_snapshot
|
||||
runtime = SimpleNamespace(context=context)
|
||||
return _FakeRequest(
|
||||
runtime=runtime,
|
||||
tools=[_FakeTool(name=name) for name in tools],
|
||||
system_message=SystemMessage(content=[{"type": "text", "text": "base"}]),
|
||||
state=state or {},
|
||||
)
|
||||
|
||||
|
||||
def _build_tool_request(*, skills: list[str], visible_skills: list[str], file_path: str) -> _FakeToolCallRequest:
|
||||
return _FakeToolCallRequest(
|
||||
tool_call={"name": "read_file", "args": {"file_path": file_path}},
|
||||
runtime=SimpleNamespace(
|
||||
context=SimpleNamespace(
|
||||
skills=skills,
|
||||
skill_session_snapshot=_build_snapshot(selected=skills, visible=visible_skills),
|
||||
)
|
||||
),
|
||||
state={},
|
||||
)
|
||||
|
||||
|
||||
def _extract_appended_prompt(request: _FakeRequest) -> str:
|
||||
return request.system_message.content_blocks[-1]["text"]
|
||||
|
||||
|
||||
def _build_middleware() -> RuntimeConfigMiddleware:
|
||||
return RuntimeConfigMiddleware(
|
||||
enable_model_override=False,
|
||||
enable_tools_override=False,
|
||||
enable_system_prompt_override=True,
|
||||
enable_skills_prompt_override=True,
|
||||
)
|
||||
|
||||
|
||||
def _build_snapshot(
|
||||
selected: list[str],
|
||||
visible: list[str] | None = None,
|
||||
metadata: dict[str, dict[str, str]] | None = None,
|
||||
dependency_map: dict[str, dict[str, list[str]]] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"selected_skills": selected,
|
||||
"visible_skills": visible if visible is not None else selected,
|
||||
"prompt_metadata": metadata or {},
|
||||
"dependency_map": dependency_map or {},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_abefore_agent_resolves_visible_skills_and_preinjects_prompt(monkeypatch: pytest.MonkeyPatch):
|
||||
async def fake_resolve(selected):
|
||||
assert selected == ["deliver-prd"]
|
||||
return _build_snapshot(
|
||||
selected=["deliver-prd"],
|
||||
visible=["deliver-prd", "brainstorming"],
|
||||
metadata={
|
||||
"deliver-prd": {
|
||||
"name": "deliver-prd",
|
||||
"description": "deliver prd",
|
||||
"path": "/skills/deliver-prd/SKILL.md",
|
||||
},
|
||||
"brainstorming": {
|
||||
"name": "brainstorming",
|
||||
"description": "brainstorming desc",
|
||||
"path": "/skills/brainstorming/SKILL.md",
|
||||
},
|
||||
},
|
||||
dependency_map={
|
||||
"deliver-prd": {"tools": [], "mcps": [], "skills": ["brainstorming"]},
|
||||
"brainstorming": {"tools": [], "mcps": [], "skills": []},
|
||||
},
|
||||
)
|
||||
|
||||
monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
|
||||
middleware = _build_middleware()
|
||||
request = _build_request(skills=["deliver-prd"], tools=["read_file"], system_prompt="你是助手")
|
||||
|
||||
await middleware.abefore_agent(request.state, request.runtime)
|
||||
|
||||
snapshot = request.runtime.context.skill_session_snapshot
|
||||
assert snapshot["selected_skills"] == ["deliver-prd"]
|
||||
assert snapshot["visible_skills"] == ["deliver-prd", "brainstorming"]
|
||||
assert snapshot["dependency_map"]["deliver-prd"]["skills"] == ["brainstorming"]
|
||||
assert request.runtime.context._skills_prompt_injected is True
|
||||
assert "## Skills System" in request.runtime.context.system_prompt
|
||||
assert "- **deliver-prd**: deliver prd" in request.runtime.context.system_prompt
|
||||
assert "- **brainstorming**: brainstorming desc" in request.runtime.context.system_prompt
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_abefore_agent_injection_is_idempotent(monkeypatch: pytest.MonkeyPatch):
|
||||
async def fake_resolve(_selected):
|
||||
return _build_snapshot(
|
||||
selected=["alpha"],
|
||||
metadata={"alpha": {"name": "alpha", "description": "alpha desc", "path": "/skills/alpha/SKILL.md"}},
|
||||
)
|
||||
|
||||
monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
|
||||
middleware = _build_middleware()
|
||||
request = _build_request(skills=["alpha"], tools=["read_file"], system_prompt="base prompt")
|
||||
|
||||
await middleware.abefore_agent(request.state, request.runtime)
|
||||
await middleware.abefore_agent(request.state, request.runtime)
|
||||
|
||||
assert request.runtime.context.system_prompt.count("## Skills System") == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_awrap_model_call_keeps_preinjected_skills_prompt(monkeypatch: pytest.MonkeyPatch):
|
||||
middleware = _build_middleware()
|
||||
request = _build_request(
|
||||
skills=["alpha"],
|
||||
tools=["read_file"],
|
||||
system_prompt="base prompt\n\n## Skills System\n- **alpha**: alpha desc",
|
||||
)
|
||||
|
||||
def raise_if_called(_skills_meta):
|
||||
raise AssertionError("_build_skills_section should not be called in awrap_model_call")
|
||||
|
||||
monkeypatch.setattr(middleware, "_build_skills_section", raise_if_called)
|
||||
result = await middleware.awrap_model_call(request, _echo_handler)
|
||||
prompt = _extract_appended_prompt(result)
|
||||
|
||||
assert "当前时间：" in prompt
|
||||
assert "## Skills System" in prompt
|
||||
assert "- **alpha**: alpha desc" in prompt
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_abefore_agent_degrades_when_resolver_fails(monkeypatch: pytest.MonkeyPatch):
|
||||
async def fake_resolve(_selected):
|
||||
raise RuntimeError("boom")
|
||||
|
||||
monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
|
||||
middleware = _build_middleware()
|
||||
request = _build_request(skills=["alpha"], tools=["read_file"], system_prompt="base prompt")
|
||||
|
||||
await middleware.abefore_agent(request.state, request.runtime)
|
||||
|
||||
snapshot = request.runtime.context.skill_session_snapshot
|
||||
assert snapshot["selected_skills"] == ["alpha"]
|
||||
assert snapshot["visible_skills"] == []
|
||||
assert "## Skills System" not in request.runtime.context.system_prompt
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_awrap_tool_call_activates_skill_when_visible_in_context():
|
||||
middleware = _build_middleware()
|
||||
request = _build_tool_request(
|
||||
skills=["deliver-prd"],
|
||||
visible_skills=["deliver-prd", "brainstorming"],
|
||||
file_path="/skills/brainstorming/SKILL.md",
|
||||
)
|
||||
|
||||
async def _handler(_request):
|
||||
return ToolMessage(content="ok", tool_call_id="tc-1")
|
||||
|
||||
result = await middleware.awrap_tool_call(request, _handler)
|
||||
assert isinstance(result, Command)
|
||||
assert result.update["activated_skills"] == ["brainstorming"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_awrap_tool_call_denies_invisible_skill():
|
||||
middleware = _build_middleware()
|
||||
request = _build_tool_request(
|
||||
skills=["deliver-prd"],
|
||||
visible_skills=["deliver-prd"],
|
||||
file_path="/skills/brainstorming/SKILL.md",
|
||||
)
|
||||
|
||||
async def _handler(_request):
|
||||
return ToolMessage(content="ok", tool_call_id="tc-1")
|
||||
|
||||
result = await middleware.awrap_tool_call(request, _handler)
|
||||
assert isinstance(result, ToolMessage)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_model_call_injects_dependency_tools_and_mcps_after_activation(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(runtime_middleware, "get_kb_based_tools", lambda db_names=None: [])
|
||||
|
||||
snapshot = _build_snapshot(
|
||||
selected=["alpha"],
|
||||
visible=["alpha", "beta"],
|
||||
dependency_map={"alpha": {"tools": ["dep-tool"], "mcps": ["mcp-a"], "skills": ["beta"]}},
|
||||
)
|
||||
|
||||
def fake_build_dependency_bundle(received_snapshot, activated):
|
||||
assert received_snapshot == snapshot
|
||||
assert activated == ["alpha"]
|
||||
return {"tools": ["dep-tool"], "mcps": ["mcp-a"], "skills": ["alpha", "beta"]}
|
||||
|
||||
monkeypatch.setattr(runtime_middleware, "build_dependency_bundle", fake_build_dependency_bundle)
|
||||
|
||||
async def fake_get_enabled_mcp_tools(server_name: str):
|
||||
if server_name == "mcp-a":
|
||||
return [_FakeTool(name="mcp_tool")]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(runtime_middleware, "get_enabled_mcp_tools", fake_get_enabled_mcp_tools)
|
||||
|
||||
middleware = RuntimeConfigMiddleware(
|
||||
extra_tools=[_FakeTool(name="mcp_tool")],
|
||||
enable_model_override=False,
|
||||
enable_tools_override=True,
|
||||
enable_system_prompt_override=False,
|
||||
enable_skills_prompt_override=False,
|
||||
)
|
||||
|
||||
request = _build_request(
|
||||
skills=["alpha"],
|
||||
tools=["calculator", "dep-tool", "mcp_tool", "read_file"],
|
||||
state={"activated_skills": ["alpha"]},
|
||||
skill_snapshot=snapshot,
|
||||
)
|
||||
|
||||
result = await middleware.awrap_model_call(request, _echo_handler)
|
||||
tool_names = [t.name for t in result.tools]
|
||||
assert "dep-tool" in tool_names
|
||||
assert "mcp_tool" in tool_names
|
||||
assert "calculator" not in tool_names
|
||||
@ -1,85 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from src.services import skill_resolver as resolver
|
||||
from src.storage.postgres.models_business import Skill
|
||||
|
||||
|
||||
def test_expand_skill_closure_and_dependency_bundle():
|
||||
dependency_map = {
|
||||
"alpha": {"tools": ["t1"], "mcps": ["m1"], "skills": ["beta"]},
|
||||
"beta": {"tools": ["t2"], "mcps": ["m2"], "skills": ["gamma"]},
|
||||
"gamma": {"tools": ["t3"], "mcps": [], "skills": []},
|
||||
}
|
||||
snapshot: resolver.SkillSessionSnapshot = {
|
||||
"selected_skills": ["alpha"],
|
||||
"visible_skills": ["alpha", "beta", "gamma"],
|
||||
"prompt_metadata": {},
|
||||
"dependency_map": dependency_map,
|
||||
}
|
||||
|
||||
closure = resolver.expand_skill_closure(["alpha"], dependency_map)
|
||||
assert closure == ["alpha", "beta", "gamma"]
|
||||
|
||||
bundle = resolver.build_dependency_bundle(snapshot, ["alpha"])
|
||||
assert bundle["skills"] == ["alpha", "beta", "gamma"]
|
||||
assert bundle["tools"] == ["t1", "t2", "t3"]
|
||||
assert bundle["mcps"] == ["m1", "m2"]
|
||||
|
||||
|
||||
def test_expand_skill_closure_cycle():
|
||||
dependency_map = {
|
||||
"alpha": {"tools": [], "mcps": [], "skills": ["beta"]},
|
||||
"beta": {"tools": [], "mcps": [], "skills": ["alpha"]},
|
||||
}
|
||||
assert resolver.expand_skill_closure(["alpha"], dependency_map) == ["alpha", "beta"]
|
||||
|
||||
|
||||
def test_collect_prompt_metadata_order_and_dedup():
|
||||
snapshot: resolver.SkillSessionSnapshot = {
|
||||
"selected_skills": ["beta", "alpha"],
|
||||
"visible_skills": ["beta", "alpha"],
|
||||
"prompt_metadata": {
|
||||
"beta": {"name": "beta", "description": "beta skill", "path": "/skills/beta/SKILL.md"},
|
||||
"alpha": {"name": "alpha", "description": "alpha skill", "path": "/skills/alpha/SKILL.md"},
|
||||
},
|
||||
"dependency_map": {},
|
||||
}
|
||||
result = resolver.collect_prompt_metadata(snapshot, ["beta", "missing", "alpha", "beta"])
|
||||
assert [item["name"] for item in result] == ["beta", "alpha"]
|
||||
assert [item["path"] for item in result] == ["/skills/beta/SKILL.md", "/skills/alpha/SKILL.md"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_session_snapshot_and_selected_change(monkeypatch: pytest.MonkeyPatch):
|
||||
async def fake_list_skills(_db=None):
|
||||
return [
|
||||
Skill(
|
||||
slug="alpha",
|
||||
name="alpha",
|
||||
description="a",
|
||||
tool_dependencies=[],
|
||||
mcp_dependencies=[],
|
||||
skill_dependencies=["beta"],
|
||||
dir_path="skills/alpha",
|
||||
),
|
||||
Skill(
|
||||
slug="beta",
|
||||
name="beta",
|
||||
description="b",
|
||||
tool_dependencies=[],
|
||||
mcp_dependencies=[],
|
||||
skill_dependencies=[],
|
||||
dir_path="skills/beta",
|
||||
),
|
||||
]
|
||||
|
||||
monkeypatch.setattr(resolver, "_list_skills_from_db", fake_list_skills)
|
||||
|
||||
snapshot = await resolver.resolve_session_snapshot([" alpha ", "alpha"])
|
||||
assert snapshot["selected_skills"] == ["alpha"]
|
||||
assert snapshot["visible_skills"] == ["alpha", "beta"]
|
||||
|
||||
assert resolver.is_snapshot_match_selected_skills(snapshot, ["alpha"]) is True
|
||||
assert resolver.is_snapshot_match_selected_skills(snapshot, ["beta"]) is False
|
||||
@ -7,6 +7,7 @@ from pathlib import Path
|
||||
import pytest
|
||||
|
||||
from src.services import skill_service as svc
|
||||
from src.services import tool_service
|
||||
from src.storage.postgres.models_business import Skill
|
||||
|
||||
|
||||
@ -31,15 +32,26 @@ def test_parse_skill_markdown_requires_frontmatter():
|
||||
svc._parse_skill_markdown("# missing")
|
||||
|
||||
|
||||
def test_validate_skill_slug():
|
||||
assert svc.validate_skill_slug("demo-skill") == "demo-skill"
|
||||
with pytest.raises(ValueError, match="无效 skill slug"):
|
||||
svc.validate_skill_slug("../bad")
|
||||
def test_is_valid_skill_slug():
|
||||
# Test valid slugs
|
||||
assert svc.is_valid_skill_slug("demo-skill") is True
|
||||
assert svc.is_valid_skill_slug("valid-name-123") is True
|
||||
# Test invalid slugs
|
||||
assert svc.is_valid_skill_slug("../bad") is False
|
||||
assert svc.is_valid_skill_slug("Invalid") is False # uppercase not allowed
|
||||
assert svc.is_valid_skill_slug("") is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_skill_dependency_options(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(svc, "_get_buildin_tool_names", lambda: ["calculator", "search"])
|
||||
# Mock get_tool_metadata to return tool list
|
||||
def fake_get_tool_metadata(category=None):
|
||||
return [
|
||||
{"id": "calculator", "name": "Calculator"},
|
||||
{"id": "search", "name": "Search"},
|
||||
]
|
||||
|
||||
monkeypatch.setattr(tool_service, "get_tool_metadata", fake_get_tool_metadata)
|
||||
monkeypatch.setattr(svc, "get_mcp_server_names", lambda: ["mcp-a", "mcp-b"])
|
||||
|
||||
class FakeRepo:
|
||||
@ -55,7 +67,7 @@ async def test_get_skill_dependency_options(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(svc, "SkillRepository", FakeRepo)
|
||||
|
||||
result = await svc.get_skill_dependency_options(None)
|
||||
assert result["tools"] == ["calculator", "search"]
|
||||
assert result["tools"] == [{"id": "calculator", "name": "Calculator"}, {"id": "search", "name": "Search"}]
|
||||
assert result["mcps"] == ["mcp-a", "mcp-b"]
|
||||
assert result["skills"] == ["alpha", "beta"]
|
||||
|
||||
@ -202,7 +214,12 @@ async def test_update_skill_dependencies(monkeypatch: pytest.MonkeyPatch):
|
||||
mcp_dependencies=[],
|
||||
skill_dependencies=[],
|
||||
)
|
||||
monkeypatch.setattr(svc, "_get_buildin_tool_names", lambda: ["calculator"])
|
||||
|
||||
# Mock get_tool_metadata to return tool list
|
||||
def fake_get_tool_metadata(category=None):
|
||||
return [{"id": "calculator", "name": "Calculator"}]
|
||||
|
||||
monkeypatch.setattr(tool_service, "get_tool_metadata", fake_get_tool_metadata)
|
||||
monkeypatch.setattr(svc, "get_mcp_server_names", lambda: ["mcp-a"])
|
||||
|
||||
async def fake_get_skill_or_raise(_db, slug: str):
|
||||
@ -262,4 +279,4 @@ async def test_update_skill_dependencies(monkeypatch: pytest.MonkeyPatch):
|
||||
assert captured["mcp_dependencies"] == ["mcp-a"]
|
||||
assert captured["skill_dependencies"] == ["beta"]
|
||||
assert captured["updated_by"] == "root"
|
||||
assert updated.skill_dependencies == ["beta"]
|
||||
assert updated.skill_dependencies == ["beta"]
|
||||
@ -1,7 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
from src.agents.common.backends import skills_backend
|
||||
|
||||
@ -43,70 +42,4 @@ def test_selected_skills_backend_readonly_and_visible_only_selected(tmp_path, mo
|
||||
|
||||
upload_result = backend.upload_files([("/alpha/a.txt", b"a")])
|
||||
assert len(upload_result) == 1
|
||||
assert upload_result[0].error == "permission_denied"
|
||||
|
||||
|
||||
def test_composite_backend_mounts_skills_under_prefix(tmp_path, monkeypatch):
|
||||
_prepare_skills_dir(tmp_path)
|
||||
monkeypatch.setattr(skills_backend, "get_skills_root_dir", lambda: tmp_path)
|
||||
|
||||
runtime = SimpleNamespace(
|
||||
context=SimpleNamespace(
|
||||
skills=["alpha"],
|
||||
skill_session_snapshot={
|
||||
"selected_skills": ["alpha"],
|
||||
"visible_skills": ["alpha", "beta"],
|
||||
"prompt_metadata": {},
|
||||
"dependency_map": {},
|
||||
},
|
||||
),
|
||||
state={},
|
||||
)
|
||||
composite = skills_backend.create_agent_composite_backend(runtime)
|
||||
|
||||
root = composite.ls_info("/")
|
||||
all_paths = [entry.get("path") for entry in root]
|
||||
assert "/skills/" in all_paths
|
||||
|
||||
skills_root = composite.ls_info("/skills/")
|
||||
skill_paths = sorted(entry.get("path") for entry in skills_root)
|
||||
assert skill_paths == ["/skills/alpha/", "/skills/beta/"]
|
||||
|
||||
denied = composite.write("/skills/alpha/new.md", "x")
|
||||
assert denied.error and "read-only" in denied.error
|
||||
|
||||
|
||||
def test_composite_backend_fallbacks_to_context_skills_when_snapshot_missing(tmp_path, monkeypatch):
|
||||
_prepare_skills_dir(tmp_path)
|
||||
monkeypatch.setattr(skills_backend, "get_skills_root_dir", lambda: tmp_path)
|
||||
|
||||
runtime = SimpleNamespace(
|
||||
context=SimpleNamespace(skills=["alpha"]),
|
||||
state={},
|
||||
)
|
||||
composite = skills_backend.create_agent_composite_backend(runtime)
|
||||
skills_root = composite.ls_info("/skills/")
|
||||
skill_paths = sorted(entry.get("path") for entry in skills_root)
|
||||
assert skill_paths == ["/skills/alpha/"]
|
||||
|
||||
|
||||
def test_composite_backend_reads_visible_skills_from_runtime_context(tmp_path, monkeypatch):
|
||||
_prepare_skills_dir(tmp_path)
|
||||
monkeypatch.setattr(skills_backend, "get_skills_root_dir", lambda: tmp_path)
|
||||
|
||||
runtime = SimpleNamespace(
|
||||
context=SimpleNamespace(
|
||||
skills=["alpha"],
|
||||
skill_session_snapshot={
|
||||
"selected_skills": ["alpha"],
|
||||
"visible_skills": ["alpha", "beta"],
|
||||
"prompt_metadata": {},
|
||||
"dependency_map": {},
|
||||
},
|
||||
),
|
||||
state=None,
|
||||
)
|
||||
composite = skills_backend.create_agent_composite_backend(runtime)
|
||||
skills_root = composite.ls_info("/skills/")
|
||||
skill_paths = sorted(entry.get("path") for entry in skills_root)
|
||||
assert skill_paths == ["/skills/alpha/", "/skills/beta/"]
|
||||
assert upload_result[0].error == "permission_denied"
|
||||
Loading…
Reference in New Issue
Block a user