q792602257andClaude Opus 4.6
cbfdb2ae39
Adds the skills:admin router (cloud/sdk/skill_api.py) for cloud-skill CRUD and per-host entitlement grant/revoke with CSRF/scope/audit, and a host-scoped router serving incremental per-host sync deltas plus the agent local-skill inventory report/readback. Both composed into the Cloud API app. cloud-api suite green (46 passed). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
129 lines
4.3 KiB
Python
129 lines
4.3 KiB
Python
"""HTTP tests for the Cloud skill management admin router.
|
|
|
|
Mirrors the llm-provider management test setup (in-memory DB, admin login,
|
|
CSRF). Covers skill CRUD, per-host entitlement grant/revoke, authorization
|
|
(non-admin rejected), and a basic sync-endpoint auth guard.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from cloud.control_config import CloudControlConfig
|
|
from cloud_api.app import create_app
|
|
|
|
|
|
def _create_admin(client: TestClient) -> None:
|
|
client.app.state.cloud_services.user_auth_service.create_user(
|
|
username="admin",
|
|
display_name="Administrator",
|
|
role="admin",
|
|
password="correct-horse-battery-staple",
|
|
must_change_password=False,
|
|
)
|
|
response = client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "admin", "password": "correct-horse-battery-staple"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def _csrf_headers(client: TestClient) -> dict[str, str]:
|
|
token = client.cookies.get("amcp_csrf")
|
|
assert token is not None
|
|
return {"X-CSRF-Token": token}
|
|
|
|
|
|
def _skill_payload(**overrides: object) -> dict[str, object]:
|
|
payload: dict[str, object] = {
|
|
"name": "Search Notes",
|
|
"kind": "knowledge",
|
|
"description": "how to search",
|
|
"tags": ["search"],
|
|
"content": "type and press enter",
|
|
"steps": [],
|
|
"parameters": {},
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _client() -> TestClient:
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
return TestClient(app)
|
|
|
|
|
|
def test_admin_can_create_list_get_update_delete_skill():
|
|
with _client() as client:
|
|
_create_admin(client)
|
|
headers = _csrf_headers(client)
|
|
|
|
created = client.post("/v1/skills", json=_skill_payload(), headers=headers)
|
|
assert created.status_code == 201, created.text
|
|
skill_id = created.json()["id"]
|
|
|
|
listed = client.get("/v1/skills", headers=headers)
|
|
assert listed.status_code == 200
|
|
assert any(s["id"] == skill_id for s in listed.json()["items"])
|
|
|
|
fetched = client.get(f"/v1/skills/{skill_id}", headers=headers)
|
|
assert fetched.status_code == 200
|
|
assert fetched.json()["content"] == "type and press enter"
|
|
|
|
updated = client.patch(
|
|
f"/v1/skills/{skill_id}",
|
|
json=_skill_payload(content="new content"),
|
|
headers=headers,
|
|
)
|
|
assert updated.status_code == 200, updated.text
|
|
assert updated.json()["content"] == "new content"
|
|
|
|
deleted = client.delete(f"/v1/skills/{skill_id}", headers=headers)
|
|
assert deleted.status_code == 204
|
|
assert client.get(f"/v1/skills/{skill_id}", headers=headers).status_code == 404
|
|
|
|
|
|
def test_duplicate_skill_name_conflicts():
|
|
with _client() as client:
|
|
_create_admin(client)
|
|
headers = _csrf_headers(client)
|
|
first = client.post("/v1/skills", json=_skill_payload(), headers=headers)
|
|
assert first.status_code == 201
|
|
second = client.post("/v1/skills", json=_skill_payload(), headers=headers)
|
|
assert second.status_code == 409
|
|
|
|
|
|
def test_entitlement_grant_revoke_lists_hosts():
|
|
with _client() as client:
|
|
_create_admin(client)
|
|
headers = _csrf_headers(client)
|
|
skill_id = client.post(
|
|
"/v1/skills", json=_skill_payload(), headers=headers
|
|
).json()["id"]
|
|
|
|
grant = client.post(
|
|
f"/v1/skills/{skill_id}/entitlements/host-1", headers=headers
|
|
)
|
|
assert grant.status_code == 204
|
|
listed = client.get(f"/v1/skills/{skill_id}/entitlements", headers=headers)
|
|
assert listed.json()["host_ids"] == ["host-1"]
|
|
|
|
revoke = client.delete(
|
|
f"/v1/skills/{skill_id}/entitlements/host-1", headers=headers
|
|
)
|
|
assert revoke.status_code == 204
|
|
listed = client.get(f"/v1/skills/{skill_id}/entitlements", headers=headers)
|
|
assert listed.json()["host_ids"] == []
|
|
|
|
|
|
def test_unauthenticated_request_is_rejected():
|
|
with _client() as client:
|
|
response = client.get("/v1/skills")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_sync_endpoint_requires_host_credentials():
|
|
with _client() as client:
|
|
# No host credentials -> 401 (no skill content leaked).
|
|
response = client.get("/internal/v1/hosts/host-1/skills/sync")
|
|
assert response.status_code == 401
|