agent-platform/tests/test_skill_service_api.py

from pathlib import Path

from tests.conftest import (
    build_fastapi_test_client,
    build_postgres_database_url,
    build_postgres_engine,
    prepare_known_service_import,
)


def test_skill_service_post_contract_matches_frontend(
    tmp_path: Path,
    monkeypatch,
) -> None:
    prepare_known_service_import("skill-service")

    from app.bootstrap.app import create_app
    from app.db.models import Base
    from core_db import create_session_factory

    database_url = build_postgres_database_url(tmp_path, "skill-api")
    monkeypatch.setenv("AGENT_PLATFORM_DATABASE_URL", database_url)
    engine = build_postgres_engine(database_url)
    Base.metadata.create_all(engine)

    app = create_app()
    app.state.session_factory = create_session_factory(engine)
    client = build_fastapi_test_client(app)

    create_response = client.post(
        "/skills/create",
        json={
            "name": "订单查询",
            "description": "查询订单并组织回复",
            "category": "service",
            "instruction": "使用订单工具查询状态，然后回复客户。",
            "toolIds": ["tool_lookup_order"],
        },
    )
    assert create_response.status_code == 200
    skill = create_response.json()["data"]
    assert skill["name"] == "订单查询"
    assert skill["category"] == "service"
    assert skill["toolIds"] == ["tool_lookup_order"]
    assert skill["createdTime"]
    assert "code" not in skill
    assert "skill_type" not in skill

    list_response = client.post(
        "/skills/list",
        json={"page": 1, "pageSize": 20, "keyword": "订单"},
    )
    assert list_response.status_code == 200
    list_payload = list_response.json()["data"]
    assert list_payload["total"] == 1
    assert list_payload["items"][0]["id"] == skill["id"]

    update_response = client.post(
        "/skills/update",
        json={
            "skillId": skill["id"],
            "name": "订单状态查询",
            "instruction": "先查询订单，再解释当前状态。",
            "toolIds": ["tool_lookup_order", "tool_create_ticket"],
        },
    )
    assert update_response.status_code == 200
    updated = update_response.json()["data"]
    assert updated["name"] == "订单状态查询"
    assert updated["toolIds"] == ["tool_lookup_order", "tool_create_ticket"]

    assert skill["runtimeType"] == "template"
    assert skill["implementation"]["template"]

    install_response = client.post(
        "/skills/installations/create",
        json={
            "skillId": skill["id"],
            "installScope": "global",
            "scopeId": "global",
            "config": {"priority": "normal"},
        },
    )
    assert install_response.status_code == 200
    installation = install_response.json()["data"]
    assert installation["skillId"] == skill["id"]
    assert installation["status"] == "installed"

    installations_response = client.post(
        "/skills/installations/list",
        json={"page": 1, "pageSize": 20, "installScope": "global"},
    )
    assert installations_response.status_code == 200
    assert installations_response.json()["data"]["total"] == 1

    run_create_response = client.post(
        "/skills/runs/create",
        json={
            "skillId": skill["id"],
            "installationId": installation["id"],
            "input": {"orderNo": "SO-1001"},
        },
    )
    assert run_create_response.status_code == 200
    run = run_create_response.json()["data"]
    assert run["skillId"] == skill["id"]
    assert run["installationId"] == installation["id"]
    assert run["input"] == {"orderNo": "SO-1001"}
    assert "skill_id" not in run
    assert "input_json" not in run

    run_execute_response = client.post(
        "/skills/runs/execute",
        json={"skillRunId": run["id"], "workerKey": "pytest-worker"},
    )
    assert run_execute_response.status_code == 200
    executed_run = run_execute_response.json()["data"]
    assert executed_run["status"] == "completed"
    assert executed_run["workerKey"] == "pytest-worker"
    assert executed_run["finishedTime"]
    assert "worker_key" not in executed_run

    delete_response = client.post("/skills/delete", json={"skillId": skill["id"]})
    assert delete_response.status_code == 200
    assert delete_response.json()["data"] == {"deleted": True, "skillId": skill["id"], "installationId": None}