fischer-agentkit/tests/unit/test_skill_loader_provenance.py

"""SkillLoader v7 provenance + 危险能力告警单元测试"""

import os
import tempfile
from unittest.mock import patch

import yaml

from agentkit.skills.base import Skill, SkillConfig
from agentkit.skills.loader import SkillLoader
from agentkit.skills.registry import SkillRegistry


def _write_yaml(directory: str, filename: str, data: dict) -> str:
    path = os.path.join(directory, filename)
    with open(path, "w", encoding="utf-8") as f:
        yaml.dump(data, f, allow_unicode=True)
    return path


class _FakeEntryPoint:
    """模拟 importlib.metadata.EntryPoint"""

    def __init__(self, name: str, skill: Skill):
        self.name = name
        self._skill = skill

    def load(self):
        return self._skill


def _make_skill(name: str = "ep_skill", capabilities=None, tools=None) -> Skill:
    config = SkillConfig(
        name=name,
        agent_type="test",
        task_mode="llm_generate",
        prompt={"identity": "test"},
        capabilities=capabilities,
        tools=tools,
    )
    return Skill(config)


class TestSkillLoaderProvenance:
    def test_load_from_file_sets_yaml_provenance(self):
        registry = SkillRegistry()
        loader = SkillLoader(skill_registry=registry)
        with tempfile.TemporaryDirectory() as tmpdir:
            path = _write_yaml(
                tmpdir,
                "s.yaml",
                {
                    "name": "s",
                    "agent_type": "t",
                    "task_mode": "llm_generate",
                    "prompt": {"identity": "x"},
                },
            )
            skill = loader.load_from_file(path)
            assert skill.config.provenance == f"yaml:{path}"

    def test_load_from_skill_md_sets_provenance(self):
        registry = SkillRegistry()
        loader = SkillLoader(skill_registry=registry)
        skill_md = """\
---
name: md-skill
description: "test"
agent_type: test
execution_mode: react
---

# Trigger
- test

# Steps
1. step

# Pitfalls
- none

# Verification
- ok
"""
        with tempfile.TemporaryDirectory() as tmpdir:
            path = os.path.join(tmpdir, "SKILL.md")
            with open(path, "w", encoding="utf-8") as f:
                f.write(skill_md)
            skill = loader.load_from_skill_md(path)
            assert skill.config.provenance == f"skill_md:{path}"

    def test_load_from_entry_points_sets_provenance(self):
        registry = SkillRegistry()
        loader = SkillLoader(skill_registry=registry)
        fake_ep = _FakeEntryPoint("my_ep", _make_skill("ep_skill"))
        with patch("agentkit.skills.loader.sys.version_info", (3, 12, 0)):
            with patch("importlib.metadata.entry_points", return_value=[fake_ep]):
                skills = loader.load_from_entry_points()
        assert len(skills) == 1
        assert skills[0].config.provenance == "entry_point:my_ep"

    def test_entry_points_dangerous_capability_warning(self, caplog):
        """entry_points 加载声明 shell 能力的 Skill 时触发 warning"""
        import logging

        registry = SkillRegistry()
        loader = SkillLoader(skill_registry=registry)
        dangerous_skill = _make_skill(
            "dangerous_skill", capabilities=[{"tag": "shell"}, {"tag": "code_execution"}]
        )
        fake_ep = _FakeEntryPoint("dangerous_ep", dangerous_skill)
        with patch("agentkit.skills.loader.sys.version_info", (3, 12, 0)):
            with patch("importlib.metadata.entry_points", return_value=[fake_ep]):
                with caplog.at_level(logging.WARNING):
                    skills = loader.load_from_entry_points()
        assert len(skills) == 1
        assert skills[0].config.provenance == "entry_point:dangerous_ep"
        # warning 包含 skill 名与危险能力
        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
        assert any(
            "dangerous_skill" in r.getMessage() and "shell" in r.getMessage() for r in warnings
        )

    def test_entry_points_dangerous_tools_warning(self, caplog):
        """entry_points 加载绑定 shell 工具但未声明 capabilities 的 Skill 时触发 warning"""
        import logging

        registry = SkillRegistry()
        loader = SkillLoader(skill_registry=registry)
        # 有危险 tools 但无 capabilities 声明——旧逻辑会漏检
        dangerous_skill = _make_skill("stealthy_skill", capabilities=None, tools=["shell"])
        fake_ep = _FakeEntryPoint("stealthy_ep", dangerous_skill)
        with patch("agentkit.skills.loader.sys.version_info", (3, 12, 0)):
            with patch("importlib.metadata.entry_points", return_value=[fake_ep]):
                with caplog.at_level(logging.WARNING):
                    skills = loader.load_from_entry_points()
        assert len(skills) == 1
        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
        assert any(
            "stealthy_skill" in r.getMessage() and "shell" in r.getMessage() for r in warnings
        )

    def test_entry_points_no_capabilities_no_warning(self, caplog):
        import logging

        registry = SkillRegistry()
        loader = SkillLoader(skill_registry=registry)
        safe_skill = _make_skill("safe_skill", capabilities=None)
        fake_ep = _FakeEntryPoint("safe_ep", safe_skill)
        with patch("agentkit.skills.loader.sys.version_info", (3, 12, 0)):
            with patch("importlib.metadata.entry_points", return_value=[fake_ep]):
                with caplog.at_level(logging.WARNING):
                    skills = loader.load_from_entry_points()
        assert len(skills) == 1
        # 不应有危险能力 warning（只可能有其他 warning）
        dangerous_warnings = [
            r
            for r in caplog.records
            if r.levelno == logging.WARNING and "dangerous capabilities" in r.getMessage()
        ]
        assert dangerous_warnings == []

    def test_yaml_provenance_overridden_by_loader(self):
        """YAML 中已有 provenance 字段时，加载路径覆盖它（加载路径是权威来源）"""
        registry = SkillRegistry()
        loader = SkillLoader(skill_registry=registry)
        with tempfile.TemporaryDirectory() as tmpdir:
            path = _write_yaml(
                tmpdir,
                "s.yaml",
                {
                    "name": "s",
                    "agent_type": "t",
                    "task_mode": "llm_generate",
                    "prompt": {"identity": "x"},
                    "provenance": "user_supplied:should_be_overridden",
                },
            )
            skill = loader.load_from_file(path)
            assert skill.config.provenance == f"yaml:{path}"
            assert "user_supplied" not in skill.config.provenance