251 lines
9.1 KiB
Python
251 lines
9.1 KiB
Python
"""Tests for the AI assistant skills & slash-commands (#81)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import backend.skills as skills_mod
|
|
from backend.skills import (
|
|
BUILTIN_SKILLS,
|
|
create_user_skill,
|
|
delete_user_skill,
|
|
get_skill_prompt,
|
|
list_skills,
|
|
list_user_skills,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def skill_store(tmp_path, monkeypatch):
|
|
"""Point the skills store at an isolated temp file."""
|
|
store = tmp_path / "skills.json"
|
|
monkeypatch.setattr(skills_mod, "SKILLS_FILE", store)
|
|
return store
|
|
|
|
|
|
USER = {"username": "alice"}
|
|
|
|
|
|
class TestBuiltinSkills:
|
|
def test_expected_skills_present(self):
|
|
ids = {s["id"] for s in BUILTIN_SKILLS}
|
|
expected = {
|
|
# Base
|
|
"research", "create-new-skill", "resume", "actions", "reformuler",
|
|
"correction", "brainstorm", "plan", "ask", "meeting-note", "livrable",
|
|
# Extraction & structuration
|
|
"extract", "timeline", "glossary", "tag",
|
|
# Transformation & adaptation
|
|
"translate", "adapt", "clean", "summary-progressive",
|
|
# Analyse critique & décision
|
|
"critique", "compare", "prioritize", "swot", "debate",
|
|
# Apprentissage & mémorisation
|
|
"quiz", "reading-note", "qa-generator",
|
|
# Méta-gestion & confidentialité
|
|
"link", "anonymize", "estimate",
|
|
}
|
|
assert expected <= ids
|
|
|
|
def test_ids_unique_and_valid(self):
|
|
ids = [s["id"] for s in BUILTIN_SKILLS]
|
|
assert len(ids) == len(set(ids))
|
|
for skill_id in ids:
|
|
assert skills_mod._SKILL_ID_RE.match(skill_id), skill_id
|
|
|
|
def test_every_skill_has_label_and_description(self):
|
|
for skill in BUILTIN_SKILLS:
|
|
assert skill["label"]
|
|
assert skill["description"]
|
|
assert skill.get("icon")
|
|
|
|
def test_every_skill_has_prompt_with_common_rules(self):
|
|
for skill in BUILTIN_SKILLS:
|
|
assert skill.get("prompt"), skill["id"]
|
|
assert skills_mod.COMMON_RULES in skill["prompt"], skill["id"]
|
|
|
|
def test_builtin_prompt_resolves(self, skill_store):
|
|
prompt = get_skill_prompt("research", USER)
|
|
assert prompt and "analyste de recherche documentaire" in prompt
|
|
|
|
def test_unknown_skill_returns_none(self, skill_store):
|
|
assert get_skill_prompt("does-not-exist", USER) is None
|
|
|
|
def test_list_includes_admin_commands(self, skill_store):
|
|
data = list_skills(USER)
|
|
command_ids = {c["id"] for c in data["commands"]}
|
|
assert {"help", "providers", "provider", "model", "keys"} <= command_ids
|
|
|
|
|
|
class TestUserSkills:
|
|
def test_create_and_list(self, skill_store):
|
|
skill = create_user_skill(USER, {
|
|
"id": "my-skill", "label": "Mon skill", "prompt": "Fais X",
|
|
})
|
|
assert skill["id"] == "my-skill"
|
|
assert skill["custom"] is True
|
|
stored = list_user_skills(USER)
|
|
assert [s["id"] for s in stored] == ["my-skill"]
|
|
assert skill_store.exists()
|
|
|
|
def test_create_rejects_reserved_id(self, skill_store):
|
|
with pytest.raises(ValueError):
|
|
create_user_skill(USER, {"id": "research", "label": "x", "prompt": "y"})
|
|
|
|
def test_create_rejects_duplicate(self, skill_store):
|
|
create_user_skill(USER, {"id": "dup", "label": "x", "prompt": "y"})
|
|
with pytest.raises(ValueError):
|
|
create_user_skill(USER, {"id": "dup", "label": "x", "prompt": "y"})
|
|
|
|
def test_create_rejects_bad_id(self, skill_store):
|
|
with pytest.raises(ValueError):
|
|
create_user_skill(USER, {"id": "Bad ID!", "label": "x", "prompt": "y"})
|
|
|
|
def test_create_requires_prompt(self, skill_store):
|
|
with pytest.raises(ValueError):
|
|
create_user_skill(USER, {"id": "noprompt", "label": "x", "prompt": ""})
|
|
|
|
def test_user_skill_prompt_resolves(self, skill_store):
|
|
create_user_skill(USER, {"id": "custom", "label": "C", "prompt": "Instruction Z"})
|
|
assert get_skill_prompt("custom", USER) == "Instruction Z"
|
|
|
|
def test_skills_are_per_user(self, skill_store):
|
|
create_user_skill(USER, {"id": "only-alice", "label": "A", "prompt": "p"})
|
|
assert list_user_skills({"username": "bob"}) == []
|
|
|
|
def test_delete(self, skill_store):
|
|
create_user_skill(USER, {"id": "temp", "label": "T", "prompt": "p"})
|
|
assert delete_user_skill(USER, "temp") is True
|
|
assert list_user_skills(USER) == []
|
|
assert delete_user_skill(USER, "temp") is False
|
|
|
|
def test_corrupted_store_is_ignored(self, skill_store):
|
|
skill_store.write_text("{not json", encoding="utf-8")
|
|
assert list_user_skills(USER) == []
|
|
# Creating still works (store is rewritten).
|
|
create_user_skill(USER, {"id": "recover", "label": "R", "prompt": "p"})
|
|
data = json.loads(skill_store.read_text(encoding="utf-8"))
|
|
assert data["alice"][0]["id"] == "recover"
|
|
|
|
|
|
# ── Endpoint tests ───────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def skills_client(tmp_path, monkeypatch):
|
|
"""TestClient with auth enabled and an isolated data dir."""
|
|
import asyncio
|
|
import shutil
|
|
import tempfile
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
tmp = Path(tempfile.mkdtemp())
|
|
data_dir = tmp / "data"
|
|
data_dir.mkdir()
|
|
|
|
from backend.auth.password import hash_password
|
|
users = {
|
|
"version": 1,
|
|
"users": {
|
|
"testuser": {
|
|
"id": "testuser-1",
|
|
"username": "testuser",
|
|
"display_name": "Test User",
|
|
"password_hash": hash_password("TestPass123!"),
|
|
"role": "admin",
|
|
"vaults": ["*"],
|
|
"active": True,
|
|
"created_at": "2026-01-01T00:00:00",
|
|
}
|
|
},
|
|
}
|
|
(data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")
|
|
src_secret = Path("data/secret.key")
|
|
if src_secret.exists():
|
|
shutil.copy2(str(src_secret), str(data_dir / "secret.key"))
|
|
|
|
orig_cwd = os.getcwd()
|
|
os.chdir(str(tmp))
|
|
monkeypatch.setattr(skills_mod, "SKILLS_FILE", Path("data/skills.json"))
|
|
|
|
os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
|
|
os.environ["OBSIGATE_ADMIN_USER"] = "testuser"
|
|
os.environ["OBSIGATE_ADMIN_PASSWORD"] = "TestPass123!"
|
|
os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
|
|
os.environ["VAULT_1_NAME"] = "TestVault"
|
|
os.environ["VAULT_1_PATH"] = str(tmp / "vault")
|
|
|
|
import backend.main
|
|
backend.main._load_config = lambda: {"watcher_enabled": False}
|
|
from backend.indexer import index
|
|
for key in list(index.keys()):
|
|
del index[key]
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
client = TestClient(backend.main.app)
|
|
yield client
|
|
client.close()
|
|
os.chdir(orig_cwd)
|
|
shutil.rmtree(str(tmp), ignore_errors=True)
|
|
for k in ["OBSIGATE_AUTH_ENABLED", "OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD",
|
|
"OBSIGATE_WATCHER_ENABLED", "VAULT_1_NAME", "VAULT_1_PATH"]:
|
|
os.environ.pop(k, None)
|
|
|
|
|
|
def _token(client):
|
|
resp = client.post("/api/auth/login", json={"username": "testuser", "password": "TestPass123!"})
|
|
return resp.json()["access_token"]
|
|
|
|
|
|
class TestSkillsEndpoints:
|
|
def test_list_requires_auth(self, skills_client):
|
|
assert skills_client.get("/api/ai/skills").status_code == 401
|
|
|
|
def test_list_returns_builtins_and_commands(self, skills_client):
|
|
token = _token(skills_client)
|
|
resp = skills_client.get("/api/ai/skills", headers={"Authorization": f"Bearer {token}"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
ids = {s["id"] for s in data["skills"]}
|
|
assert "research" in ids
|
|
assert any(c["id"] == "help" for c in data["commands"])
|
|
|
|
def test_create_and_delete_roundtrip(self, skills_client):
|
|
token = _token(skills_client)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = skills_client.post(
|
|
"/api/ai/skills",
|
|
json={"id": "custom", "label": "Custom", "prompt": "Do it"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["id"] == "custom"
|
|
|
|
listed = skills_client.get("/api/ai/skills", headers=headers).json()
|
|
assert "custom" in {s["id"] for s in listed["skills"]}
|
|
|
|
deleted = skills_client.delete("/api/ai/skills/custom", headers=headers)
|
|
assert deleted.status_code == 200
|
|
|
|
def test_create_invalid_id_returns_400(self, skills_client):
|
|
token = _token(skills_client)
|
|
resp = skills_client.post(
|
|
"/api/ai/skills",
|
|
json={"id": "Bad!", "label": "x", "prompt": "y"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_delete_unknown_returns_404(self, skills_client):
|
|
token = _token(skills_client)
|
|
resp = skills_client.delete(
|
|
"/api/ai/skills/unknown", headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
assert resp.status_code == 404
|