X-Financial/server/tests/test_agent_asset_service.py

from __future__ import annotations

import uuid

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.core.agent_enums import (
    AgentAssetContentType,
    AgentAssetDomain,
    AgentAssetStatus,
    AgentAssetType,
    AgentName,
    AgentReviewStatus,
    AgentRunSource,
    AgentRunStatus,
)
from app.db.base import Base
from app.schemas.agent_asset import (
    AgentAssetCreate,
    AgentAssetReviewCreate,
    AgentAssetVersionCreate,
)
from app.services.agent_assets import AgentAssetService
from app.services.agent_runs import AgentRunService
from app.services.audit import AuditLogService


def build_session() -> Session:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    return session_factory()


def test_agent_asset_service_seeds_assets_and_enforces_review_before_activation() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        rules = service.list_assets(asset_type=AgentAssetType.RULE.value)
        assert len(rules) >= 3
        assert any(
            item.code == "rule.expense.travel_risk_control_standard" and item.status == AgentAssetStatus.ACTIVE.value
            for item in rules
        )
        assert all(
            item.code
            not in {
                "rule.expense.duplicate_expense_check",
                "rule.expense.travel_receipt_requirements",
                "rule.ap.payment_dual_review",
            }
            for item in rules
        )

        pending_rule = next(item for item in rules if item.status == AgentAssetStatus.REVIEW.value)

        with pytest.raises(PermissionError):
            service.activate_asset(pending_rule.id, actor="pytest")


def test_agent_asset_service_seeds_all_foundation_asset_types() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        assert len(service.list_assets(asset_type=AgentAssetType.RULE.value)) >= 3
        assert len(service.list_assets(asset_type=AgentAssetType.SKILL.value)) >= 2
        assert len(service.list_assets(asset_type=AgentAssetType.MCP.value)) >= 2
        assert len(service.list_assets(asset_type=AgentAssetType.TASK.value)) >= 3


def test_agent_asset_service_can_activate_rule_after_review() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        created = service.create_asset(
            AgentAssetCreate(
                asset_type=AgentAssetType.RULE,
                code=f"rule.test.{uuid.uuid4().hex[:8]}",
                name="测试规则",
                description="用于测试审核和上线流程。",
                domain=AgentAssetDomain.EXPENSE,
                scenario_json=["expense", "risk_check"],
                owner="pytest",
                reviewer="reviewer",
                status=AgentAssetStatus.DRAFT,
                config_json={"enabled": False},
            ),
            actor="pytest",
        )
        service.create_version(
            created.id,
            AgentAssetVersionCreate(
                version="v1.0.0",
                content="# 测试规则\n\n- 仅用于测试。",
                content_type=AgentAssetContentType.MARKDOWN,
                change_note="初始化版本",
                created_by="pytest",
            ),
            actor="pytest",
        )
        service.create_review(
            created.id,
            AgentAssetReviewCreate(
                version="v1.0.0",
                reviewer="reviewer",
                review_status=AgentReviewStatus.APPROVED,
                review_note="可以上线",
            ),
            actor="reviewer",
        )

        activated = service.activate_asset(created.id, actor="reviewer")

        assert activated.status == AgentAssetStatus.ACTIVE.value
        assert activated.current_version == "v1.0.0"
        assert activated.latest_review is not None
        assert activated.latest_review.review_status == AgentReviewStatus.APPROVED.value


def test_agent_asset_service_returns_recent_versions_for_rule_detail() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.attachment_submission_requirements"
        )
        detail = service.get_asset(rule.id)

        assert detail is not None
        assert detail.current_version == "v1.0.0"
        assert detail.current_version_content_type == AgentAssetContentType.MARKDOWN.value
        assert isinstance(detail.current_version_content, str)
        assert len(detail.recent_versions) >= 2
        assert any(item.is_current for item in detail.recent_versions)
        assert {item.version for item in detail.recent_versions} >= {"v0.9.0", "v1.0.0"}
        assert detail.config_json["rule_template_key"] == "attachment_requirement_v1"
        assert "附件或单据不完整" in str(detail.current_version_content)


def test_agent_asset_service_returns_travel_policy_rule_detail() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.travel_risk_control_standard"
        )
        detail = service.get_asset(rule.id)

        assert detail is not None
        assert detail.status == AgentAssetStatus.ACTIVE.value
        assert detail.current_version == "v1.1.0"
        assert detail.latest_review is not None
        assert detail.latest_review.review_status == AgentReviewStatus.APPROVED.value
        assert "行程闭环" in str(detail.current_version_content)
        assert "住宿标准、飞机舱位和火车席别" in str(detail.current_version_content)


def test_agent_run_service_lists_seeded_trace_data() -> None:
    with build_session() as db:
        service = AgentRunService(db)

        runs = service.list_runs()

        assert len(runs) >= 3
        assert any(item.tool_calls for item in runs)
        assert any(item.semantic_parse is not None for item in runs)


def test_agent_run_service_creates_run_and_persists_error_message() -> None:
    with build_session() as db:
        service = AgentRunService(db)

        created = service.create_run(
            agent=AgentName.ORCHESTRATOR.value,
            source=AgentRunSource.SYSTEM_EVENT.value,
            status=AgentRunStatus.FAILED.value,
            error_message="simulated failure",
            result_summary="failed to route request",
        )
        fetched = service.get_run(created.run_id)

        assert fetched is not None
        assert fetched.run_id.startswith("run_")
        assert fetched.status == AgentRunStatus.FAILED.value
        assert fetched.error_message == "simulated failure"
        assert fetched.result_summary == "failed to route request"


def test_agent_asset_creation_writes_audit_log() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        created = service.create_asset(
            AgentAssetCreate(
                asset_type=AgentAssetType.SKILL,
                code=f"skill.test.{uuid.uuid4().hex[:8]}",
                name="测试技能",
                description="用于测试审计日志写入。",
                domain=AgentAssetDomain.KNOWLEDGE,
                scenario_json=["knowledge", "query"],
                owner="pytest",
                reviewer="reviewer",
                status=AgentAssetStatus.DRAFT,
                config_json={"enabled": True},
            ),
            actor="pytest",
        )
        logs = AuditLogService(db).list_logs(resource_id=created.id)

        assert any(item.action == "create_agent_asset" for item in logs)