bruno
55696bfb31
CI / lint (push) Successful in 57s
CI / security (push) Successful in 39s
CI / test (push) Successful in 1m13s
CI / build (push) Successful in 36s
CI / e2e (push) Successful in 10m13s
Desktop Build / build-windows (push) Canceled after 0s
Desktop Build / build-linux (push) Canceled after 0s
- backend/ai_chat.py: chat_completion provider-agnostique (OpenAI-compat tools/tool_calls + Gemini functionDeclarations/functionCall), retry sans tools si rejete - backend/agent/loop.py: run_agent multi-etapes (limite 10, truncation, confirmation two-step), LLM injectable - endpoint opt-in POST /api/ai/bookslm/agent (events SSE tool/message/confirmation), extraction _resolve_system_prompt - tests: test_ai_chat.py, test_agent_loop.py + 3 tests endpoint (728 passed au total) - ROADMAP B1/B2/B3/B7 livres ; B4/B5/B6 restants
826 lines
30 KiB
Python
826 lines
30 KiB
Python
"""Tests for BooksLM — directory context collection, caching, and API routes."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
# ── Unit tests: collect_directory_context ──────────────────────────────
|
|
|
|
|
|
class TestCollectDirectoryContext:
|
|
"""Tests for collect_directory_context()."""
|
|
|
|
def test_basic_collection(self, tmp_path):
|
|
"""Collect .md files from a simple directory."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "note1.md").write_text("# Note 1\nHello world", encoding="utf-8")
|
|
(vault / "note2.md").write_text("# Note 2\nGoodbye world", encoding="utf-8")
|
|
(vault / "notemd.txt").write_text("Not markdown", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
assert result["file_count"] == 2
|
|
assert result["total_chars"] > 0
|
|
paths = [f["path"] for f in result["files"]]
|
|
assert "note1.md" in paths
|
|
assert "note2.md" in paths
|
|
assert "notemd.txt" not in paths
|
|
|
|
def test_subdirectory_collection(self, tmp_path):
|
|
"""Collect files recursively from subdirectories."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
subdir = vault / "projects" / "code"
|
|
subdir.mkdir(parents=True)
|
|
(subdir / "readme.md").write_text("# Code project", encoding="utf-8")
|
|
(vault / "root.md").write_text("# Root", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "projects")
|
|
|
|
assert result["file_count"] == 1
|
|
assert result["files"][0]["path"] == "projects/code/readme.md"
|
|
|
|
def test_hidden_files_skipped(self, tmp_path):
|
|
"""Hidden files and special directories are skipped."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / ".hidden.md").write_text("Hidden", encoding="utf-8")
|
|
(vault / ".obsidian").mkdir()
|
|
(vault / ".obsidian" / "config.md").write_text("Config", encoding="utf-8")
|
|
(vault / "visible.md").write_text("Visible", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
assert result["file_count"] == 1
|
|
assert result["files"][0]["path"] == "visible.md"
|
|
|
|
def test_attachments_skipped(self, tmp_path):
|
|
"""_attachments/ directory is skipped."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
attach = vault / "_attachments"
|
|
attach.mkdir(parents=True)
|
|
(attach / "image.md").write_text("Image doc", encoding="utf-8")
|
|
(vault / "real.md").write_text("Real content", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
assert result["file_count"] == 1
|
|
assert result["files"][0]["path"] == "real.md"
|
|
|
|
def test_max_files_limit(self, tmp_path):
|
|
"""Respects BOOKSLM_MAX_FILES limit."""
|
|
from backend.bookslm import collect_directory_context
|
|
import backend.bookslm as bookslm_mod
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
|
|
old_max = bookslm_mod.BOOKSLM_MAX_FILES
|
|
bookslm_mod.BOOKSLM_MAX_FILES = 3
|
|
try:
|
|
for i in range(10):
|
|
(vault / f"note{i}.md").write_text(f"Content {i}", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
assert result["file_count"] == 3
|
|
finally:
|
|
bookslm_mod.BOOKSLM_MAX_FILES = old_max
|
|
|
|
def test_max_file_chars_truncation(self, tmp_path):
|
|
"""Files exceeding max chars are truncated with marker."""
|
|
from backend.bookslm import collect_directory_context
|
|
import backend.bookslm as bookslm_mod
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
|
|
old_max = bookslm_mod.BOOKSLM_MAX_FILE_CHARS
|
|
bookslm_mod.BOOKSLM_MAX_FILE_CHARS = 50
|
|
try:
|
|
long_content = "x" * 200
|
|
(vault / "long.md").write_text(long_content, encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
assert result["file_count"] == 1
|
|
assert "[... tronqué]" in result["files"][0]["content"]
|
|
assert len(result["files"][0]["content"]) <= 50 + 50 # truncated + marker
|
|
finally:
|
|
bookslm_mod.BOOKSLM_MAX_FILE_CHARS = old_max
|
|
|
|
def test_max_total_chars_limit(self, tmp_path):
|
|
"""Stops collecting when total chars limit is reached."""
|
|
from backend.bookslm import collect_directory_context
|
|
import backend.bookslm as bookslm_mod
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
|
|
old_total = bookslm_mod.BOOKSLM_MAX_TOTAL_CHARS
|
|
old_file = bookslm_mod.BOOKSLM_MAX_FILE_CHARS
|
|
bookslm_mod.BOOKSLM_MAX_TOTAL_CHARS = 100
|
|
bookslm_mod.BOOKSLM_MAX_FILE_CHARS = 10000
|
|
try:
|
|
for i in range(10):
|
|
(vault / f"f{i}.md").write_text("a" * 50, encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
# Should not collect all 10 files (10 * 50 = 500 > 100)
|
|
assert result["total_chars"] <= 100 + 50 # some margin for truncation marker
|
|
finally:
|
|
bookslm_mod.BOOKSLM_MAX_TOTAL_CHARS = old_total
|
|
bookslm_mod.BOOKSLM_MAX_FILE_CHARS = old_file
|
|
|
|
def test_readme_index_priority(self, tmp_path):
|
|
"""README and index files come first in results."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "aaa.md").write_text("# AAA", encoding="utf-8")
|
|
(vault / "README.md").write_text("# README", encoding="utf-8")
|
|
(vault / "index.md").write_text("# Index", encoding="utf-8")
|
|
(vault / "zzz.md").write_text("# ZZZ", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
paths = [f["path"] for f in result["files"]]
|
|
# README and index should be before other files
|
|
readme_idx = paths.index("README.md")
|
|
index_idx = paths.index("index.md")
|
|
aaa_idx = paths.index("aaa.md")
|
|
assert readme_idx < aaa_idx
|
|
assert index_idx < aaa_idx
|
|
|
|
def test_empty_directory(self, tmp_path):
|
|
"""Empty directory returns empty result."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
assert result["file_count"] == 0
|
|
assert result["total_chars"] == 0
|
|
assert result["files"] == []
|
|
|
|
def test_nonexistent_directory(self, tmp_path):
|
|
"""Nonexistent directory returns empty result."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
result = collect_directory_context(tmp_path, "nonexistent")
|
|
|
|
assert result["file_count"] == 0
|
|
|
|
def test_title_generation(self, tmp_path):
|
|
"""File titles are derived from stem with proper casing."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "my-cool-note.md").write_text("Content", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
assert result["files"][0]["title"] == "My Cool Note"
|
|
|
|
def test_directory_tree(self, tmp_path):
|
|
"""Directory tree is included in result."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
(vault / "sub").mkdir(parents=True)
|
|
(vault / "sub" / "file.md").write_text("Content", encoding="utf-8")
|
|
(vault / "root.md").write_text("Root", encoding="utf-8")
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
tree = result["directory_tree"]
|
|
assert "root.md" in tree
|
|
assert "sub/" in tree
|
|
assert "file.md" in tree
|
|
|
|
|
|
# ── Unit tests: build_system_prompt ────────────────────────────────────
|
|
|
|
|
|
class TestBuildSystemPrompt:
|
|
"""Tests for build_system_prompt()."""
|
|
|
|
def test_basic_prompt(self):
|
|
"""Prompt contains expected sections."""
|
|
from backend.bookslm import build_system_prompt
|
|
|
|
context = {
|
|
"files": [
|
|
{"path": "note.md", "title": "My Note", "content": "# Hello", "type": "markdown"},
|
|
],
|
|
"total_chars": 7,
|
|
"file_count": 1,
|
|
"directory_tree": "note.md",
|
|
}
|
|
|
|
prompt = build_system_prompt(context)
|
|
|
|
assert "assistant de recherche" in prompt
|
|
assert "note.md" in prompt
|
|
assert "My Note" in prompt
|
|
assert "# Hello" in prompt
|
|
assert "Cite tes sources" in prompt
|
|
|
|
def test_empty_context(self):
|
|
"""Empty context still produces valid prompt."""
|
|
from backend.bookslm import build_system_prompt
|
|
|
|
context = {"files": [], "total_chars": 0, "file_count": 0, "directory_tree": ""}
|
|
prompt = build_system_prompt(context)
|
|
|
|
assert "0 fichier" in prompt
|
|
|
|
def test_token_warning(self):
|
|
"""Large context triggers token warning."""
|
|
from backend.bookslm import build_system_prompt
|
|
|
|
context = {
|
|
"files": [
|
|
{"path": "big.md", "title": "Big", "content": "x" * 500000, "type": "markdown"},
|
|
],
|
|
"total_chars": 500000,
|
|
"file_count": 1,
|
|
"directory_tree": "big.md",
|
|
}
|
|
|
|
prompt = build_system_prompt(context)
|
|
|
|
assert "⚠️" in prompt or "volumineux" in prompt
|
|
|
|
|
|
# ── Unit tests: caching ────────────────────────────────────────────────
|
|
|
|
|
|
class TestCaching:
|
|
"""Tests for cache behavior."""
|
|
|
|
def test_cache_hit(self, tmp_path):
|
|
"""Second call with same data returns cached result."""
|
|
from backend.bookslm import collect_directory_context, _cache
|
|
|
|
_cache.clear()
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "note.md").write_text("Content", encoding="utf-8")
|
|
|
|
result1 = collect_directory_context(vault, "")
|
|
result2 = collect_directory_context(vault, "")
|
|
|
|
assert result1["file_count"] == result2["file_count"]
|
|
assert result1["total_chars"] == result2["total_chars"]
|
|
|
|
def test_cache_invalidation_on_change(self, tmp_path):
|
|
"""Cache is invalidated when file content changes."""
|
|
from backend.bookslm import collect_directory_context, _cache
|
|
|
|
_cache.clear()
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "note.md").write_text("Original", encoding="utf-8")
|
|
|
|
result1 = collect_directory_context(vault, "")
|
|
assert result1["file_count"] == 1
|
|
|
|
# Modify file (change mtime)
|
|
import time
|
|
time.sleep(0.1)
|
|
(vault / "note.md").write_text("Modified content", encoding="utf-8")
|
|
|
|
result2 = collect_directory_context(vault, "")
|
|
assert result2["files"][0]["content"] == "Modified content"
|
|
|
|
def test_invalidate_cache(self, tmp_path):
|
|
"""invalidate_cache clears the cache."""
|
|
from backend.bookslm import collect_directory_context, invalidate_cache, _cache
|
|
|
|
_cache.clear()
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "note.md").write_text("Content", encoding="utf-8")
|
|
|
|
collect_directory_context(vault, "")
|
|
assert len(_cache) > 0
|
|
|
|
count = invalidate_cache()
|
|
assert count > 0
|
|
assert len(_cache) == 0
|
|
|
|
|
|
# ── Unit tests: secret redaction ──────────────────────────────────────
|
|
|
|
|
|
class TestRedaction:
|
|
"""Verify that secrets are redacted in collected content."""
|
|
|
|
def test_secrets_are_redacted(self, tmp_path):
|
|
"""API keys in file content are redacted."""
|
|
from backend.bookslm import collect_directory_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
# The secret redactor looks for patterns like sk-..., AKIA..., etc.
|
|
(vault / "secrets.md").write_text(
|
|
"Config: api_key=AKIA1234567890ABCDEF and sk-abcdefghijklmnopqrstuvwxyz01234567890",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = collect_directory_context(vault, "")
|
|
|
|
# The redactor should have processed this file
|
|
content = result["files"][0]["content"]
|
|
# At minimum the file should be collected (redaction is best-effort)
|
|
assert result["file_count"] == 1
|
|
|
|
|
|
# ── Integration tests: API endpoints ──────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def bookslm_client():
|
|
"""Create a TestClient with auth enabled, isolated temp data."""
|
|
tmp = Path(tempfile.mkdtemp())
|
|
data_dir = tmp / "data"
|
|
data_dir.mkdir()
|
|
|
|
# Create a test vault with some files
|
|
test_vault = tmp / "test-vault"
|
|
test_vault.mkdir()
|
|
(test_vault / "README.md").write_text("# Test Vault\nWelcome to the test vault.", encoding="utf-8")
|
|
(test_vault / "notes").mkdir()
|
|
(test_vault / "notes" / "note1.md").write_text("# Note 1\nFirst note content.", encoding="utf-8")
|
|
(test_vault / "notes" / "note2.md").write_text("# Note 2\nSecond note content.", encoding="utf-8")
|
|
|
|
from backend.auth.password import hash_password
|
|
pw_hash = hash_password("TestPass123!")
|
|
users = {
|
|
"version": 1,
|
|
"users": {
|
|
"testuser": {
|
|
"id": "testuser-1",
|
|
"username": "testuser",
|
|
"display_name": "Test User",
|
|
"password_hash": pw_hash,
|
|
"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))
|
|
|
|
os.environ["VAULT_1_NAME"] = "TestVault"
|
|
os.environ["VAULT_1_PATH"] = str(test_vault)
|
|
os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
|
|
os.environ["OBSIGATE_ADMIN_USER"] = "testuser"
|
|
os.environ["OBSIGATE_ADMIN_PASSWORD"] = "TestPass123!"
|
|
os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
|
|
|
|
import backend.main
|
|
backend.main._load_config = lambda: {"watcher_enabled": False}
|
|
|
|
from backend.main import app
|
|
from backend.indexer import build_index, index
|
|
for key in list(index.keys()):
|
|
del index[key]
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_until_complete(build_index())
|
|
|
|
from backend.search import init_inverted_index
|
|
init_inverted_index()
|
|
|
|
from fastapi.testclient import TestClient
|
|
client = TestClient(app)
|
|
yield client
|
|
|
|
if hasattr(client, 'close'):
|
|
client.close()
|
|
loop.run_until_complete(asyncio.sleep(0))
|
|
|
|
os.chdir(orig_cwd)
|
|
shutil.rmtree(str(tmp), ignore_errors=True)
|
|
for k in ["VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
|
|
"OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD", "OBSIGATE_WATCHER_ENABLED"]:
|
|
os.environ.pop(k, None)
|
|
|
|
|
|
def _login_bookslm(client, username="testuser", password="TestPass123!"):
|
|
resp = client.post("/api/auth/login", json={"username": username, "password": password})
|
|
return resp.json().get("access_token"), resp
|
|
|
|
|
|
class TestBooksLMContextEndpoint:
|
|
"""Tests for POST /api/ai/bookslm/context."""
|
|
|
|
def test_context_returns_files(self, bookslm_client):
|
|
"""Context endpoint returns files from the directory."""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"vault": "TestVault", "directory": "notes"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["file_count"] == 2
|
|
paths = [f["path"] for f in data["files"]]
|
|
assert "notes/note1.md" in paths
|
|
assert "notes/note2.md" in paths
|
|
|
|
def test_context_root_directory(self, bookslm_client):
|
|
"""Context endpoint works for root directory."""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"vault": "TestVault", "directory": ""},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["file_count"] >= 1 # At least README.md
|
|
|
|
def test_context_requires_auth(self, bookslm_client):
|
|
"""Context endpoint requires authentication."""
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"vault": "TestVault", "directory": ""},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_context_vault_not_found(self, bookslm_client):
|
|
"""Context endpoint returns 404 for unknown vault."""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"vault": "NonExistent", "directory": ""},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_context_nonexistent_directory(self, bookslm_client):
|
|
"""Context endpoint returns empty for nonexistent directory."""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"vault": "TestVault", "directory": "nonexistent"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["file_count"] == 0
|
|
|
|
|
|
class TestBooksLMChatEndpoint:
|
|
"""Tests for POST /api/ai/chat."""
|
|
|
|
def test_chat_requires_auth(self, bookslm_client):
|
|
"""Chat endpoint requires authentication."""
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/chat",
|
|
json={"vault": "TestVault", "directory": "", "message": "Hello"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_chat_vault_not_found(self, bookslm_client):
|
|
"""Chat endpoint returns 404 for unknown vault."""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/chat",
|
|
json={"vault": "NonExistent", "directory": "", "message": "Hello"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_chat_empty_directory(self, bookslm_client):
|
|
"""Chat endpoint returns 404 for empty directory."""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/chat",
|
|
json={"vault": "TestVault", "directory": "nonexistent", "message": "Hello"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_chat_request_accepts_provider_and_model(self, bookslm_client):
|
|
"""The chat endpoint schema accepts provider + model fields without rejecting.
|
|
|
|
We don't actually call the AI (would need a live key) — we just verify
|
|
the request schema is wired correctly so a missing key is the only
|
|
failure mode, not a 422 validation error.
|
|
"""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
# Build a tiny valid directory so the chat endpoint doesn't 404.
|
|
from pathlib import Path
|
|
vault_dir = Path(os.environ.get("VAULT_1_PATH", "test-vault"))
|
|
sub = vault_dir / "for_chat_test"
|
|
sub.mkdir(exist_ok=True)
|
|
(sub / "note.md").write_text("# hello\n", encoding="utf-8")
|
|
try:
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/chat",
|
|
json={
|
|
"vault": "TestVault",
|
|
"directory": "for_chat_test",
|
|
"message": "ping",
|
|
"provider": "deepseek",
|
|
"model": "deepseek-chat",
|
|
},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
# Either 200 (if a real key is configured) or 500 (no key / quota).
|
|
# Must NOT be 422 — the schema must accept the fields.
|
|
assert resp.status_code in (200, 500), (
|
|
f"unexpected status {resp.status_code}: {resp.text[:200]}"
|
|
)
|
|
assert resp.status_code != 422, (
|
|
f"schema rejected provider/model fields: {resp.text[:300]}"
|
|
)
|
|
finally:
|
|
import shutil
|
|
shutil.rmtree(sub, ignore_errors=True)
|
|
|
|
|
|
# ── Unit tests: explicit documents context + general prompt ────────────
|
|
|
|
|
|
class TestCollectFilesContext:
|
|
"""Tests for collect_files_context() (Documents mode)."""
|
|
|
|
def test_collects_only_requested_files(self, tmp_path):
|
|
from backend.bookslm import collect_files_context
|
|
|
|
vault = tmp_path / "vault"
|
|
(vault / "sub").mkdir(parents=True)
|
|
(vault / "a.md").write_text("# A", encoding="utf-8")
|
|
(vault / "b.md").write_text("# B", encoding="utf-8")
|
|
(vault / "sub" / "c.md").write_text("# C", encoding="utf-8")
|
|
|
|
result = collect_files_context(vault, ["a.md", "sub/c.md"], scope="documents")
|
|
|
|
assert result["scope"] == "documents"
|
|
assert result["file_count"] == 2
|
|
assert [f["path"] for f in result["files"]] == ["a.md", "sub/c.md"]
|
|
assert result["directory_tree"] == ""
|
|
|
|
def test_ignores_paths_outside_vault(self, tmp_path):
|
|
from backend.bookslm import collect_files_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "ok.md").write_text("# ok", encoding="utf-8")
|
|
outside = tmp_path / "outside.md"
|
|
outside.write_text("# secret", encoding="utf-8")
|
|
|
|
result = collect_files_context(vault, ["../outside.md", "ok.md"])
|
|
|
|
assert result["file_count"] == 1
|
|
assert result["files"][0]["path"] == "ok.md"
|
|
|
|
def test_deduplicates_paths(self, tmp_path):
|
|
from backend.bookslm import collect_files_context
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
(vault / "dup.md").write_text("# dup", encoding="utf-8")
|
|
|
|
result = collect_files_context(vault, ["dup.md", "dup.md"])
|
|
|
|
assert result["file_count"] == 1
|
|
|
|
def test_empty_context(self):
|
|
from backend.bookslm import empty_context
|
|
|
|
ctx = empty_context("general")
|
|
assert ctx["file_count"] == 0
|
|
assert ctx["files"] == []
|
|
assert ctx["scope"] == "general"
|
|
|
|
|
|
class TestGeneralPrompt:
|
|
def test_general_prompt_lists_vaults(self):
|
|
from backend.bookslm import build_general_system_prompt
|
|
|
|
prompt = build_general_system_prompt(["Alpha", "Beta"])
|
|
assert "Alpha" in prompt
|
|
assert "Beta" in prompt
|
|
assert "obsigate-action" in prompt
|
|
|
|
def test_documents_prompt_scope(self):
|
|
from backend.bookslm import build_system_prompt
|
|
|
|
context = {
|
|
"files": [{"path": "a.md", "title": "A", "content": "hello"}],
|
|
"file_count": 1,
|
|
"total_chars": 5,
|
|
"directory_tree": "",
|
|
}
|
|
prompt = build_system_prompt(context, scope="documents")
|
|
assert "Documents ouverts" in prompt
|
|
assert "hello" in prompt
|
|
|
|
|
|
class TestBooksLMContextModes:
|
|
"""Tests for the mode-aware /context endpoint."""
|
|
|
|
def test_general_mode_needs_no_vault(self, bookslm_client):
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"mode": "general"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["scope"] == "general"
|
|
assert data["file_count"] == 0
|
|
assert data["files"] == []
|
|
|
|
def test_documents_mode_returns_selected_files(self, bookslm_client):
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={
|
|
"mode": "documents",
|
|
"vault": "TestVault",
|
|
"context_files": ["notes/note1.md"],
|
|
},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["scope"] == "documents"
|
|
assert data["file_count"] == 1
|
|
assert data["files"][0]["path"] == "notes/note1.md"
|
|
|
|
def test_documents_mode_without_files_degrades_to_general(self, bookslm_client):
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"mode": "documents", "vault": "TestVault", "context_files": []},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["scope"] == "general"
|
|
assert data["file_count"] == 0
|
|
|
|
def test_documents_mode_requires_vault(self, bookslm_client):
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"mode": "documents", "context_files": ["notes/note1.md"]},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_unknown_mode_falls_back_to_directory(self, bookslm_client):
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/context",
|
|
json={"mode": "bogus", "vault": "TestVault", "directory": "notes"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["scope"] == "directory"
|
|
|
|
|
|
class TestBooksLMChatModes:
|
|
def test_general_chat_does_not_require_vault(self, bookslm_client):
|
|
"""General mode must not 400/404 on a missing vault (only 500 if no key)."""
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/chat",
|
|
json={"mode": "general", "message": "Bonjour"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code in (200, 500), f"unexpected {resp.status_code}: {resp.text[:200]}"
|
|
assert resp.status_code != 422
|
|
|
|
def test_documents_chat_accepts_context_files(self, bookslm_client):
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/chat",
|
|
json={
|
|
"mode": "documents",
|
|
"vault": "TestVault",
|
|
"context_files": ["notes/note1.md"],
|
|
"message": "Résume",
|
|
},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code in (200, 500), f"unexpected {resp.status_code}: {resp.text[:200]}"
|
|
assert resp.status_code != 422
|
|
|
|
|
|
# ── Tool-calling agent endpoint ───────────────────────────────────────
|
|
|
|
|
|
class TestBooksLMAgentEndpoint:
|
|
"""Tests for POST /api/ai/bookslm/agent (Phase B)."""
|
|
|
|
def test_agent_requires_auth(self, bookslm_client):
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/agent",
|
|
json={"vault": "TestVault", "directory": "", "message": "Hello"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_agent_tool_call_flow(self, bookslm_client, monkeypatch):
|
|
import backend.bookslm_routes as routes
|
|
from backend.ai_chat import LLMResponse, ToolCall
|
|
|
|
responses = [
|
|
LLMResponse(tool_calls=[ToolCall(id="1", name="list_directory", arguments={"vault": "TestVault"})]),
|
|
LLMResponse(content="Il y a des fichiers."),
|
|
]
|
|
|
|
async def fake_chat_completion(messages, **kwargs):
|
|
return responses.pop(0)
|
|
|
|
monkeypatch.setattr(routes, "chat_completion", fake_chat_completion)
|
|
monkeypatch.setattr(routes, "_resolve_provider_name", lambda requested: "deepseek")
|
|
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/agent",
|
|
json={"vault": "TestVault", "directory": "", "message": "liste les fichiers", "mode": "directory"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.text
|
|
assert "event: tool" in body
|
|
assert "list_directory" in body
|
|
assert "event: message" in body
|
|
assert "Il y a des fichiers." in body
|
|
|
|
def test_agent_pauses_for_confirmation(self, bookslm_client, monkeypatch):
|
|
import backend.bookslm_routes as routes
|
|
from backend.ai_chat import LLMResponse, ToolCall
|
|
from backend.tools import registry
|
|
from backend.tools.api import ToolRisk
|
|
from backend.tools.registry import ToolSpec
|
|
from backend.tools.schemas import ListVaultsInput
|
|
|
|
spec = ToolSpec(
|
|
name="_agent_write",
|
|
description="write for tests",
|
|
input_model=ListVaultsInput,
|
|
handler=lambda ctx, params: {"ok": True},
|
|
risk=ToolRisk.WRITE,
|
|
)
|
|
monkeypatch.setitem(registry._REGISTRY, "_agent_write", spec)
|
|
|
|
responses = [LLMResponse(tool_calls=[ToolCall(id="1", name="_agent_write", arguments={})])]
|
|
|
|
async def fake_chat_completion(messages, **kwargs):
|
|
return responses.pop(0)
|
|
|
|
monkeypatch.setattr(routes, "chat_completion", fake_chat_completion)
|
|
monkeypatch.setattr(routes, "_resolve_provider_name", lambda requested: "deepseek")
|
|
|
|
token, _ = _login_bookslm(bookslm_client)
|
|
resp = bookslm_client.post(
|
|
"/api/ai/bookslm/agent",
|
|
json={"vault": "TestVault", "directory": "", "message": "crée un fichier", "mode": "directory"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "event: confirmation" in resp.text
|
|
assert "_agent_write" in resp.text
|