68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from beaver.interfaces.web.app import create_app
|
|
from beaver.services.agent_service import AgentService
|
|
|
|
|
|
def _write_plugin(workspace: Path) -> None:
|
|
plugin_root = workspace / "plugins" / "baoyu-comic"
|
|
skill_root = plugin_root / "skills" / "baoyu-comic"
|
|
skill_root.mkdir(parents=True, exist_ok=True)
|
|
(skill_root / "SKILL.md").write_text(
|
|
"---\nname: baoyu-comic\ndescription: Comic workflow\n---\n\n# Comic\n\nDraw.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(plugin_root / "beaver.plugin.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"id": "baoyu-comic",
|
|
"name": "Baoyu Comic",
|
|
"version": "1.0.0",
|
|
"skills": [{"name": "baoyu-comic", "path": "skills/baoyu-comic"}],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_plugin_management_api_lifecycle(tmp_path: Path) -> None:
|
|
_write_plugin(tmp_path)
|
|
service = AgentService(workspace=tmp_path)
|
|
app = create_app(service=service, manage_service_lifecycle=False)
|
|
|
|
with TestClient(app) as client:
|
|
listed = client.get("/api/plugins")
|
|
enabled = client.post("/api/plugins/baoyu-comic/enable")
|
|
paused = client.post("/api/plugins/baoyu-comic/pause")
|
|
resumed = client.post("/api/plugins/baoyu-comic/resume")
|
|
disable_rejected = client.post("/api/plugins/baoyu-comic/disable", json={})
|
|
adopted = client.post("/api/plugins/baoyu-comic/skills/baoyu-comic/adopt")
|
|
synced = client.post("/api/plugins/sync")
|
|
|
|
assert listed.status_code == 200
|
|
assert listed.json()[0]["manifest_path"] == "plugins/baoyu-comic/beaver.plugin.json"
|
|
assert enabled.status_code == 200
|
|
assert enabled.json()["enabled"] is True
|
|
assert paused.json()["updates_paused"] is True
|
|
assert resumed.status_code == 200
|
|
assert disable_rejected.status_code == 400
|
|
assert adopted.status_code == 200
|
|
assert adopted.json()["skills"] == []
|
|
assert synced.status_code == 200
|
|
|
|
|
|
def test_plugin_management_api_unknown_plugin_returns_404(tmp_path: Path) -> None:
|
|
service = AgentService(workspace=tmp_path)
|
|
app = create_app(service=service, manage_service_lifecycle=False)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/plugins/missing/enable")
|
|
|
|
assert response.status_code == 404
|