skillwiki/tests/test_models.py

"""Phase 1 核心数据模型测试套件。"""

from __future__ import annotations

import json
from datetime import datetime
from typing import Any, Dict

import pytest

from skillos.models import (
    EdgeType,
    ExecutionStatus,
    ExperienceSourceType,
    ExperienceUnit,
    GraphStats,
    MetaSkillCategory,
    Skill,
    SkillEdge,
    SkillExecutionRecord,
    SkillEvaluation,
    SkillGraphNode,
    SkillImplementation,
    SkillInterface,
    SkillMetrics,
    SkillProposal,
    SkillProposalStatus,
    SkillProvenance,
    SkillState,
    SkillSubgraph,
    SkillTestCase,
    SkillType,
    TrajectoryStep,
)


# ===========================================================================
# Skill Model Tests
# ===========================================================================

class TestSkillModel:

    def test_minimal_skill_creation(self):
        skill = Skill(name="click_element")
        assert skill.name == "click_element"
        assert skill.skill_type == SkillType.ATOMIC
        assert skill.state == SkillState.DRAFT
        assert skill.version == "1.0.0"
        assert skill.skill_id  # UUID 自动生成

    def test_display_name_auto_set(self):
        skill = Skill(name="fill_form")
        assert skill.display_name == "Fill Form"

    def test_display_name_custom(self):
        skill = Skill(name="fill_form", display_name="Fill Structured Form")
        assert skill.display_name == "Fill Structured Form"

    def test_name_validation_snake_case(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            Skill(name="FillForm")  # 不是 snake_case
        with pytest.raises(ValidationError):
            Skill(name="fill-form")  # 连字符不允许
        with pytest.raises(ValidationError):
            Skill(name="123skill")  # 数字开头

    def test_name_valid_snake_case(self):
        skill = Skill(name="fill_form_v2")
        assert skill.name == "fill_form_v2"

    def test_tags_normalized(self):
        skill = Skill(name="test_skill", tags=["  Web  ", "FORM", "input"])
        assert skill.tags == ["web", "form", "input"]

    def test_meta_skill_requires_category(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            Skill(name="strategic_skill", skill_type=SkillType.STRATEGIC)

    def test_meta_skill_with_category(self):
        skill = Skill(
            name="lifecycle_manager",
            skill_type=SkillType.STRATEGIC,
            meta_category=MetaSkillCategory.LIFECYCLE,
        )
        assert skill.meta_category == MetaSkillCategory.LIFECYCLE

    def test_non_meta_skill_cannot_have_category(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            Skill(
                name="atomic_skill",
                skill_type=SkillType.ATOMIC,
                meta_category=MetaSkillCategory.LIFECYCLE,
            )

    def test_version_pattern(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            Skill(name="test_skill", version="v1.0")
        with pytest.raises(ValidationError):
            Skill(name="test_skill", version="1.0")

    def test_granularity_level_range(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            Skill(name="test_skill", granularity_level=0)
        with pytest.raises(ValidationError):
            Skill(name="test_skill", granularity_level=6)

    def test_state_transition_valid(self):
        skill = Skill(name="test_skill", state=SkillState.DRAFT)
        skill.transition_to(SkillState.VERIFIED)
        assert skill.state == SkillState.VERIFIED

    def test_state_transition_invalid(self):
        skill = Skill(name="test_skill", state=SkillState.DRAFT)
        with pytest.raises(ValueError, match="非法状态转换"):
            skill.transition_to(SkillState.RELEASED)  # 必须先 VERIFIED

    def test_state_transition_to_released_sets_timestamp(self):
        skill = Skill(name="test_skill", state=SkillState.VERIFIED)
        skill.transition_to(SkillState.RELEASED)
        assert skill.released_at is not None

    def test_state_transition_to_deprecated_sets_timestamp(self):
        skill = Skill(name="test_skill", state=SkillState.RELEASED)
        skill.transition_to(SkillState.DEPRECATED)
        assert skill.deprecated_at is not None

    def test_is_usable(self):
        skill = Skill(name="test_skill", state=SkillState.RELEASED)
        assert skill.is_usable()
        skill2 = Skill(name="test_skill2", state=SkillState.DEGRADED)
        assert skill2.is_usable()
        skill3 = Skill(name="test_skill3", state=SkillState.DRAFT)
        assert not skill3.is_usable()

    def test_bump_version_patch(self):
        skill = Skill(name="test_skill", version="1.2.3")
        skill.bump_version("patch")
        assert skill.version == "1.2.4"

    def test_bump_version_minor(self):
        skill = Skill(name="test_skill", version="1.2.3")
        skill.bump_version("minor")
        assert skill.version == "1.3.0"

    def test_bump_version_major(self):
        skill = Skill(name="test_skill", version="1.2.3")
        skill.bump_version("major")
        assert skill.version == "2.0.0"

    def test_record_execution_success(self):
        skill = Skill(name="test_skill")
        skill.record_execution(success=True, latency_ms=100.0)
        assert skill.metrics.usage_count == 1
        assert skill.metrics.success_count == 1
        assert skill.metrics.failure_count == 0
        assert skill.metrics.avg_latency_ms == 100.0

    def test_record_execution_failure(self):
        skill = Skill(name="test_skill")
        skill.record_execution(success=False, latency_ms=50.0)
        assert skill.metrics.failure_count == 1
        assert skill.metrics.success_count == 0

    def test_record_execution_avg_latency(self):
        skill = Skill(name="test_skill")
        skill.record_execution(success=True, latency_ms=100.0)
        skill.record_execution(success=True, latency_ms=200.0)
        assert skill.metrics.avg_latency_ms == pytest.approx(150.0)

    def test_success_rate_calculation(self):
        skill = Skill(name="test_skill")
        skill.record_execution(success=True, latency_ms=10.0)
        skill.record_execution(success=True, latency_ms=10.0)
        skill.record_execution(success=False, latency_ms=10.0)
        assert skill.metrics.success_rate == pytest.approx(2 / 3)

    def test_success_rate_zero_executions(self):
        skill = Skill(name="test_skill")
        assert skill.metrics.success_rate == 0.0

    def test_to_graph_node(self):
        skill = Skill(name="test_skill", state=SkillState.RELEASED)
        node = skill.to_graph_node()
        assert node["skill_id"] == skill.skill_id
        assert node["name"] == "test_skill"
        assert node["state"] == "S4"
        assert "success_rate" in node
        assert "usage_count" in node

    def test_default_evaluation_contract(self):
        skill = Skill(name="schema_v02_skill")
        assert skill.evaluation.verifier_specs == []
        assert skill.evaluation.test_case_refs == []
        assert skill.evaluation.benchmark_task_ids == []

    def test_skill_evaluation_normalizes_refs(self):
        evaluation = SkillEvaluation(
            verifier_specs=[{"type": "json_equals", "path": "output.success", "value": True}],
            test_case_refs=["  case-1  ", ""],
            benchmark_task_ids=[" task-1 ", "task-2"],
            validation_summary="passes fixed demo verifier",
        )
        assert evaluation.test_case_refs == ["case-1"]
        assert evaluation.benchmark_task_ids == ["task-1", "task-2"]
        assert evaluation.verifier_specs[0]["type"] == "json_equals"

    def test_skill_evaluation_rejects_blank_verifier_type(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            SkillEvaluation(verifier_specs=[{"type": "  "}])

    def test_skill_evaluation_accepts_backlog_alias(self):
        evaluation = SkillEvaluation.model_validate({"test_cases_refs": ["case-from-backlog"]})
        assert evaluation.test_case_refs == ["case-from-backlog"]


# ===========================================================================
# SkillInterface Tests
# ===========================================================================

class TestSkillInterface:

    def test_default_interface(self):
        iface = SkillInterface()
        assert iface.input_schema == {}
        assert iface.preconditions == []

    def test_full_interface(self):
        iface = SkillInterface(
            input_schema={"type": "object", "properties": {"x": {"type": "string"}}},
            preconditions=["条件1"],
            postconditions=["结果1"],
            side_effects=["副作用1"],
        )
        assert len(iface.preconditions) == 1
        assert len(iface.side_effects) == 1


# ===========================================================================
# SkillImplementation Tests
# ===========================================================================

class TestSkillImplementation:

    def test_code_implementation(self):
        impl = SkillImplementation(code="print('hello')")
        assert impl.code == "print('hello')"

    def test_prompt_implementation(self):
        impl = SkillImplementation(prompt_template="Do {action}")
        assert impl.prompt_template == "Do {action}"

    def test_sub_skill_implementation(self):
        impl = SkillImplementation(sub_skill_ids=["id1", "id2"])
        assert len(impl.sub_skill_ids) == 2

    def test_tool_call_implementation(self):
        impl = SkillImplementation(tool_calls=["browser.click"])
        assert impl.tool_calls == ["browser.click"]

    def test_empty_implementation_raises(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            SkillImplementation()  # 三者都为空


# ===========================================================================
# SkillMetrics Tests
# ===========================================================================

class TestSkillMetrics:

    def test_total_executions(self):
        m = SkillMetrics(success_count=7, failure_count=3)
        assert m.total_executions == 10

    def test_success_rate(self):
        m = SkillMetrics(success_count=9, failure_count=1)
        assert m.success_rate == pytest.approx(0.9)

    def test_success_rate_no_executions(self):
        m = SkillMetrics()
        assert m.success_rate == 0.0


# ===========================================================================
# Graph Model Tests
# ===========================================================================

class TestSkillEdge:

    def test_valid_edge(self):
        edge = SkillEdge(
            source_id="skill-a",
            target_id="skill-b",
            edge_type=EdgeType.DEPENDS_ON,
        )
        assert edge.source_id == "skill-a"
        assert edge.edge_type == EdgeType.DEPENDS_ON

    def test_self_loop_raises(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            SkillEdge(source_id="same", target_id="same", edge_type=EdgeType.DEPENDS_ON)

    def test_empty_id_raises(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            SkillEdge(source_id="", target_id="skill-b", edge_type=EdgeType.DEPENDS_ON)

    def test_weight_range(self):
        from pydantic import ValidationError
        with pytest.raises(ValidationError):
            SkillEdge(source_id="a", target_id="b", edge_type=EdgeType.SIMILAR_TO, weight=1.5)

    def test_to_neo4j_props(self):
        edge = SkillEdge(
            source_id="a", target_id="b",
            edge_type=EdgeType.COMPOSES_WITH,
            weight=0.8,
        )
        props = edge.to_neo4j_props()
        assert props["weight"] == 0.8
        assert "edge_id" in props


class TestSkillSubgraph:

    def _make_node(self, skill_id: str, name: str) -> SkillGraphNode:
        return SkillGraphNode(
            skill_id=skill_id,
            name=name,
            version="1.0.0",
            skill_type=SkillType.ATOMIC,
            state=SkillState.RELEASED,
        )

    def test_add_node(self):
        sg = SkillSubgraph()
        node = self._make_node("id1", "skill_a")
        sg.add_node(node)
        assert "id1" in sg.nodes

    def test_add_edge_valid(self):
        sg = SkillSubgraph()
        sg.add_node(self._make_node("id1", "skill_a"))
        sg.add_node(self._make_node("id2", "skill_b"))
        edge = SkillEdge(source_id="id1", target_id="id2", edge_type=EdgeType.DEPENDS_ON)
        sg.add_edge(edge)
        assert len(sg.edges) == 1

    def test_add_edge_missing_node_raises(self):
        sg = SkillSubgraph()
        sg.add_node(self._make_node("id1", "skill_a"))
        edge = SkillEdge(source_id="id1", target_id="id_missing", edge_type=EdgeType.DEPENDS_ON)
        with pytest.raises(ValueError):
            sg.add_edge(edge)

    def test_get_roots(self):
        sg = SkillSubgraph()
        sg.add_node(self._make_node("root", "root_skill"))
        sg.add_node(self._make_node("child", "child_skill"))
        edge = SkillEdge(source_id="child", target_id="root", edge_type=EdgeType.DEPENDS_ON)
        sg.add_edge(edge)
        roots = sg.get_roots()
        assert "child" in roots
        assert "root" not in roots

    def test_topological_sort(self):
        sg = SkillSubgraph()
        sg.add_node(self._make_node("a", "skill_a"))
        sg.add_node(self._make_node("b", "skill_b"))
        sg.add_node(self._make_node("c", "skill_c"))
        sg.add_edge(SkillEdge(source_id="b", target_id="a", edge_type=EdgeType.DEPENDS_ON))
        sg.add_edge(SkillEdge(source_id="c", target_id="b", edge_type=EdgeType.DEPENDS_ON))
        order = sg.topological_sort()
        assert order.index("a") < order.index("b") < order.index("c")

    def test_topological_sort_cycle_raises(self):
        sg = SkillSubgraph()
        sg.add_node(self._make_node("a", "skill_a"))
        sg.add_node(self._make_node("b", "skill_b"))
        sg.add_edge(SkillEdge(source_id="a", target_id="b", edge_type=EdgeType.DEPENDS_ON))
        sg.add_edge(SkillEdge(source_id="b", target_id="a", edge_type=EdgeType.DEPENDS_ON))
        with pytest.raises(ValueError, match="环"):
            sg.topological_sort()

    def test_to_dict(self):
        sg = SkillSubgraph(name="test_subgraph")
        sg.add_node(self._make_node("id1", "skill_a"))
        d = sg.to_dict()
        assert d["name"] == "test_subgraph"
        assert d["node_count"] == 1
        assert d["edge_count"] == 0


# ===========================================================================
# Experience Model Tests
# ===========================================================================

class TestTrajectoryStep:

    def test_valid_step(self):
        step = TrajectoryStep(
            step_index=0,
            action_type="click",
            action_target="#submit-btn",
        )
        assert step.step_index == 0
        assert step.success is True

    def test_step_with_state(self):
        step = TrajectoryStep(
            step_index=1,
            action_type="type",
            action_value="hello",
            state_before={"form_filled": False},
            state_after={"form_filled": True},
        )
        assert step.state_after["form_filled"] is True


class TestExperienceUnit:

    def test_minimal_experience(self):
        exp = ExperienceUnit(source_type=ExperienceSourceType.BROWSER_TRAJECTORY)
        assert not exp.is_processed
        assert exp.step_count == 0

    def test_experience_with_steps(self):
        steps = [
            TrajectoryStep(step_index=i, action_type="click")
            for i in range(3)
        ]
        exp = ExperienceUnit(
            source_type=ExperienceSourceType.BROWSER_TRAJECTORY,
            steps=steps,
        )
        assert exp.step_count == 3

    def test_mark_processed(self):
        exp = ExperienceUnit(source_type=ExperienceSourceType.DOCUMENTATION)
        exp.mark_processed(["skill-id-1", "skill-id-2"])
        assert exp.is_processed
        assert exp.processed_at is not None
        assert len(exp.extracted_skill_ids) == 2

    def test_tags_normalized(self):
        exp = ExperienceUnit(
            source_type=ExperienceSourceType.MANUAL_INPUT,
            tags=["  WEB  ", "FORM"],
        )
        assert exp.tags == ["web", "form"]


class TestSkillProposal:

    def test_pending_proposal(self):
        proposal = SkillProposal(
            source_experience_id="exp-123",
            proposed_name="fill_form",
            proposed_description="填写表单",
        )
        assert proposal.status == SkillProposalStatus.PENDING

    def test_accept_proposal(self):
        proposal = SkillProposal(
            source_experience_id="exp-123",
            proposed_name="fill_form",
            proposed_description="填写表单",
        )
        proposal.accept("skill-456")
        assert proposal.status == SkillProposalStatus.ACCEPTED
        assert proposal.generated_skill_id == "skill-456"

    def test_reject_proposal(self):
        proposal = SkillProposal(
            source_experience_id="exp-123",
            proposed_name="fill_form",
            proposed_description="填写表单",
        )
        proposal.reject("与已有 Skill 重复")
        assert proposal.status == SkillProposalStatus.REJECTED
        assert "重复" in proposal.rejection_reason

    def test_merge_proposal(self):
        proposal = SkillProposal(
            source_experience_id="exp-123",
            proposed_name="fill_form",
            proposed_description="填写表单",
        )
        proposal.merge_into("existing-skill-id")
        assert proposal.status == SkillProposalStatus.MERGED
        assert proposal.merged_into_skill_id == "existing-skill-id"


class TestSkillExecutionRecord:

    def test_start_execution(self):
        record = SkillExecutionRecord(skill_id="skill-1", skill_version="1.0.0")
        record.start()
        assert record.status == ExecutionStatus.RUNNING
        assert record.started_at is not None

    def test_complete_execution(self):
        record = SkillExecutionRecord(skill_id="skill-1", skill_version="1.0.0")
        record.start()
        record.complete({"result": "ok"}, {"state": "done"})
        assert record.status == ExecutionStatus.SUCCESS
        assert record.output_data == {"result": "ok"}
        assert record.latency_ms is not None
        assert record.latency_ms >= 0

    def test_fail_execution(self):
        record = SkillExecutionRecord(skill_id="skill-1", skill_version="1.0.0")
        record.start()
        record.fail("元素未找到", "ElementNotFoundError")
        assert record.status == ExecutionStatus.FAILED
        assert record.error_type == "ElementNotFoundError"
        assert record.latency_ms is not None


# ===========================================================================
# ORM Mapper Tests (no DB required)
# ===========================================================================

class TestORMMappers:

    def test_skill_to_orm_and_back(self):
        from skillos.storage.postgres_db import orm_to_skill, skill_to_orm

        original = Skill(
            name="test_skill",
            version="2.1.0",
            description="测试 Skill",
            skill_type=SkillType.FUNCTIONAL,
            domain="web",
            state=SkillState.RELEASED,
            tags=["web", "test"],
            interface=SkillInterface(
                input_schema={"type": "object"},
                preconditions=["条件1"],
            ),
            implementation=SkillImplementation(
                code="print('test')",
                tool_calls=["playwright"],
            ),
        )

        orm = skill_to_orm(original)
        assert orm.skill_id == original.skill_id
        assert orm.name == "test_skill"
        assert orm.skill_type == "functional"
        assert orm.state == "S4"
        assert json.loads(orm.tags) == ["web", "test"]

        restored = orm_to_skill(orm)
        assert restored.skill_id == original.skill_id
        assert restored.name == original.name
        assert restored.skill_type == SkillType.FUNCTIONAL
        assert restored.state == SkillState.RELEASED
        assert restored.tags == ["web", "test"]
        assert restored.interface.preconditions == ["条件1"]
        assert restored.implementation is not None
        assert restored.implementation.code == "print('test')"

    def test_skill_to_orm_no_implementation(self):
        from skillos.storage.postgres_db import skill_to_orm

        skill = Skill(name="no_impl_skill")
        orm = skill_to_orm(skill)
        assert orm.implementation_json is None

    def test_skill_to_orm_with_metrics(self):
        from skillos.storage.postgres_db import orm_to_skill, skill_to_orm

        skill = Skill(name="metrics_skill")
        skill.record_execution(success=True, latency_ms=150.0)
        skill.record_execution(success=False, latency_ms=200.0)

        orm = skill_to_orm(skill)
        assert orm.usage_count == 2
        assert orm.success_count == 1
        assert orm.failure_count == 1

        restored = orm_to_skill(orm)
        assert restored.metrics.usage_count == 2
        assert restored.metrics.success_rate == pytest.approx(0.5)