120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
from backend import cli
|
|
from backend.agents.skill_metadata import parse_skill_metadata
|
|
from backend.agents.skills_manager import SkillsManager
|
|
from backend.agents.team_pipeline_config import (
|
|
ensure_team_pipeline_config,
|
|
load_team_pipeline_config,
|
|
update_active_analysts,
|
|
)
|
|
|
|
|
|
def test_parse_skill_metadata_extended_frontmatter(tmp_path):
|
|
skill_dir = tmp_path / "demo_skill"
|
|
skill_dir.mkdir(parents=True, exist_ok=True)
|
|
(skill_dir / "SKILL.md").write_text(
|
|
"---\n"
|
|
"name: demo_skill\n"
|
|
"description: Demo description\n"
|
|
"tools:\n"
|
|
" - technical\n"
|
|
"---\n\n"
|
|
"# Demo Skill\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
parsed = parse_skill_metadata(skill_dir, source="builtin")
|
|
|
|
assert parsed.skill_name == "demo_skill"
|
|
assert parsed.description == "Demo description"
|
|
assert parsed.tools == ["technical"]
|
|
|
|
|
|
def test_update_agent_skill_overrides(tmp_path):
|
|
manager = SkillsManager(project_root=tmp_path)
|
|
asset_dir = manager.get_agent_asset_dir("demo", "risk_manager")
|
|
asset_dir.mkdir(parents=True, exist_ok=True)
|
|
(asset_dir / "agent.yaml").write_text(
|
|
"enabled_skills:\n"
|
|
" - risk_review\n"
|
|
"disabled_skills:\n"
|
|
" - old_skill\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = manager.update_agent_skill_overrides(
|
|
config_name="demo",
|
|
agent_id="risk_manager",
|
|
enable=["extra_guard"],
|
|
disable=["risk_review"],
|
|
)
|
|
|
|
assert result["enabled_skills"] == ["extra_guard"]
|
|
assert result["disabled_skills"] == ["old_skill", "risk_review"]
|
|
|
|
|
|
def test_skills_enable_disable_and_list(monkeypatch, tmp_path):
|
|
builtin_root = tmp_path / "backend" / "skills" / "builtin"
|
|
for name in ("risk_review", "extra_guard"):
|
|
skill_dir = builtin_root / name
|
|
skill_dir.mkdir(parents=True, exist_ok=True)
|
|
(skill_dir / "SKILL.md").write_text(
|
|
f"---\nname: {name}\ndescription: {name} desc\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
printed = []
|
|
monkeypatch.setattr(cli, "get_project_root", lambda: tmp_path)
|
|
monkeypatch.setattr(cli.console, "print", lambda value: printed.append(value))
|
|
|
|
cli.skills_enable(agent_id="risk_manager", skill="extra_guard", config_name="demo")
|
|
cli.skills_disable(agent_id="risk_manager", skill="risk_review", config_name="demo")
|
|
cli.skills_list(config_name="demo", agent_id="risk_manager")
|
|
|
|
text_dump = "\n".join(str(item) for item in printed)
|
|
assert "Enabled" in text_dump
|
|
assert "Disabled" in text_dump
|
|
assert any(getattr(item, "title", None) == "Skill Catalog" for item in printed)
|
|
|
|
|
|
def test_install_external_skill_for_agent(tmp_path):
|
|
manager = SkillsManager(project_root=tmp_path)
|
|
skill_dir = tmp_path / "downloaded" / "new_skill"
|
|
skill_dir.mkdir(parents=True, exist_ok=True)
|
|
(skill_dir / "SKILL.md").write_text(
|
|
"---\n"
|
|
"name: new_skill\n"
|
|
"description: external skill\n"
|
|
"---\n\n"
|
|
"# New Skill\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = manager.install_external_skill_for_agent(
|
|
config_name="demo",
|
|
agent_id="risk_manager",
|
|
source=str(skill_dir),
|
|
activate=True,
|
|
)
|
|
|
|
assert result["skill_name"] == "new_skill"
|
|
target = manager.get_agent_local_root("demo", "risk_manager") / "new_skill"
|
|
assert target.exists()
|
|
|
|
|
|
def test_team_pipeline_active_analyst_updates(tmp_path):
|
|
project_root = tmp_path
|
|
ensure_team_pipeline_config(
|
|
project_root=project_root,
|
|
config_name="demo",
|
|
default_analysts=["fundamentals_analyst", "technical_analyst"],
|
|
)
|
|
update_active_analysts(
|
|
project_root=project_root,
|
|
config_name="demo",
|
|
available_analysts=["fundamentals_analyst", "technical_analyst"],
|
|
remove=["technical_analyst"],
|
|
)
|
|
config = load_team_pipeline_config(project_root, "demo")
|
|
assert config["discussion"]["active_analysts"] == ["fundamentals_analyst"]
|