404 lines
16 KiB
Python
404 lines
16 KiB
Python
# tests/test_tools_mutations.py — Unit tests for the AI tool layer (Phase D)
|
|
"""Tests for the mutation tools: create/edit/append/rename/move/delete/restore.
|
|
|
|
These exercise the shared tool layer end-to-end against the ``TestVault``
|
|
fixture (``client``), including confirmation gating, backups and the
|
|
per-vault destructive-tools toggle.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend.tools.api import (
|
|
ToolConfirmationRequired,
|
|
ToolContext,
|
|
ToolError,
|
|
ToolPermissionError,
|
|
ToolRisk,
|
|
call_tool,
|
|
get_tool,
|
|
)
|
|
|
|
MUTATION_TOOLS = {
|
|
"create_file",
|
|
"create_directory",
|
|
"edit_file",
|
|
"append_to_file",
|
|
"rename_file",
|
|
"rename_directory",
|
|
"move_path",
|
|
"replace_in_files",
|
|
"delete_file",
|
|
"delete_directory",
|
|
"restore_backup",
|
|
}
|
|
|
|
WRITE_TOOLS = {"create_file", "create_directory", "edit_file", "append_to_file", "restore_backup"}
|
|
DANGEROUS_TOOLS = {
|
|
"rename_file",
|
|
"rename_directory",
|
|
"move_path",
|
|
"replace_in_files",
|
|
"delete_file",
|
|
"delete_directory",
|
|
}
|
|
|
|
|
|
def _ctx(vaults=None, **kwargs) -> ToolContext:
|
|
user = {"username": "tester", "role": "admin", "vaults": vaults or ["*"]}
|
|
kwargs.setdefault("audit_enabled", False)
|
|
return ToolContext(user=user, **kwargs)
|
|
|
|
|
|
def _vault_path() -> Path:
|
|
from backend.indexer import get_vault_data
|
|
|
|
return Path(get_vault_data("TestVault")["path"])
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Registry / risk levels
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestMutationRegistry:
|
|
def test_mutation_tools_registered(self):
|
|
for name in MUTATION_TOOLS:
|
|
assert get_tool(name) is not None, name
|
|
|
|
@pytest.mark.parametrize("name", sorted(WRITE_TOOLS))
|
|
def test_write_risk(self, name):
|
|
assert get_tool(name).risk is ToolRisk.WRITE
|
|
|
|
@pytest.mark.parametrize("name", sorted(DANGEROUS_TOOLS))
|
|
def test_dangerous_risk(self, name):
|
|
assert get_tool(name).risk is ToolRisk.DANGEROUS
|
|
|
|
def test_all_mutations_require_confirmation(self):
|
|
for name in MUTATION_TOOLS:
|
|
assert get_tool(name).requires_confirmation is True
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Confirmation gating
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestMutationConfirmation:
|
|
def test_create_file_requires_confirmation(self, client):
|
|
with pytest.raises(ToolConfirmationRequired):
|
|
call_tool("create_file", _ctx(), {"vault": "TestVault", "path": "n.md", "content": "x"})
|
|
assert not (_vault_path() / "n.md").exists()
|
|
|
|
def test_delete_file_requires_confirmation(self, client):
|
|
with pytest.raises(ToolConfirmationRequired):
|
|
call_tool("delete_file", _ctx(), {"vault": "TestVault", "path": "note1.md"})
|
|
assert (_vault_path() / "note1.md").exists()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# D1. Creation
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestCreate:
|
|
def test_create_file(self, client):
|
|
result = call_tool(
|
|
"create_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "new/note.md", "content": "# Hi\n"},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
assert (_vault_path() / "new" / "note.md").read_text(encoding="utf-8") == "# Hi\n"
|
|
assert result.data["path"] == "new/note.md"
|
|
|
|
def test_create_file_already_exists(self, client):
|
|
with pytest.raises(ToolError) as exc:
|
|
call_tool("create_file", _ctx(), {"vault": "TestVault", "path": "note1.md"}, confirm=True)
|
|
assert exc.value.code == "already_exists"
|
|
|
|
def test_create_file_unsupported_extension(self, client):
|
|
with pytest.raises(ToolError) as exc:
|
|
call_tool(
|
|
"create_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "binary.exe", "content": "x"},
|
|
confirm=True,
|
|
)
|
|
assert exc.value.code == "unsupported_extension"
|
|
|
|
def test_create_file_traversal_rejected(self, client):
|
|
with pytest.raises(ToolPermissionError):
|
|
call_tool(
|
|
"create_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "../evil.md", "content": "x"},
|
|
confirm=True,
|
|
)
|
|
|
|
def test_create_directory(self, client):
|
|
result = call_tool("create_directory", _ctx(), {"vault": "TestVault", "path": "A/B"}, confirm=True)
|
|
assert result.ok
|
|
assert (_vault_path() / "A" / "B").is_dir()
|
|
|
|
def test_create_directory_idempotent(self, client):
|
|
"""BUG-050: re-creating an existing folder is a success for the AI layer."""
|
|
call_tool("create_directory", _ctx(), {"vault": "TestVault", "path": "Idem/Dir"}, confirm=True)
|
|
second = call_tool("create_directory", _ctx(), {"vault": "TestVault", "path": "Idem/Dir"}, confirm=True)
|
|
assert second.ok
|
|
assert second.data.get("existed") is True
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# D2. Edit / append / rename / move
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestEditAppend:
|
|
def test_edit_file(self, client):
|
|
result = call_tool(
|
|
"edit_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "content": "# Replaced\n"},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
assert (_vault_path() / "note1.md").read_text(encoding="utf-8") == "# Replaced\n"
|
|
|
|
def test_edit_file_creates_backup(self, client):
|
|
from backend.services.backups import get_backup_dir
|
|
|
|
call_tool(
|
|
"edit_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "content": "# Changed\n"},
|
|
confirm=True,
|
|
)
|
|
backups = list(get_backup_dir("TestVault", "note1.md").glob("note1.md.*.bak"))
|
|
assert backups
|
|
|
|
def test_edit_missing_file(self, client):
|
|
with pytest.raises(ToolError) as exc:
|
|
call_tool(
|
|
"edit_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "missing.md", "content": "x"},
|
|
confirm=True,
|
|
)
|
|
assert exc.value.code == "not_found"
|
|
|
|
def test_append_to_file(self, client):
|
|
original = (_vault_path() / "note1.md").read_text(encoding="utf-8")
|
|
result = call_tool(
|
|
"append_to_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "content": "APPENDED"},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
updated = (_vault_path() / "note1.md").read_text(encoding="utf-8")
|
|
assert updated.startswith(original)
|
|
assert updated.endswith("APPENDED")
|
|
assert result.data["appended"] == len("APPENDED")
|
|
|
|
|
|
class TestRenameMove:
|
|
def test_rename_file(self, client):
|
|
result = call_tool(
|
|
"rename_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "new_name": "note1_renamed.md"},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
assert not (_vault_path() / "note1.md").exists()
|
|
assert (_vault_path() / "note1_renamed.md").exists()
|
|
assert result.data["new_path"] == "note1_renamed.md"
|
|
|
|
def test_rename_file_rejects_path_separator(self, client):
|
|
with pytest.raises(ToolError) as exc:
|
|
call_tool(
|
|
"rename_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "new_name": "sub/note1.md"},
|
|
confirm=True,
|
|
)
|
|
assert exc.value.code == "invalid_arguments"
|
|
|
|
def test_rename_directory(self, client):
|
|
call_tool("create_directory", _ctx(), {"vault": "TestVault", "path": "OldDir"}, confirm=True)
|
|
result = call_tool(
|
|
"rename_directory",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "OldDir", "new_name": "NewDir"},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
assert not (_vault_path() / "OldDir").exists()
|
|
assert (_vault_path() / "NewDir").is_dir()
|
|
|
|
def test_move_file(self, client):
|
|
result = call_tool(
|
|
"move_path",
|
|
_ctx(),
|
|
{"vault": "TestVault", "source_path": "note1.md", "destination_dir": "Projets"},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
assert result.data["new_path"] == "Projets/note1.md"
|
|
assert (_vault_path() / "Projets" / "note1.md").exists()
|
|
|
|
def test_move_directory(self, client):
|
|
result = call_tool(
|
|
"move_path",
|
|
_ctx(),
|
|
{"vault": "TestVault", "source_path": "Projets", "destination_dir": ""},
|
|
confirm=True,
|
|
)
|
|
assert result.data["item_type"] == "directory"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# D3. Find & replace
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestReplaceInFiles:
|
|
def test_dry_run_does_not_write(self, client):
|
|
before = (_vault_path() / "note1.md").read_text(encoding="utf-8")
|
|
result = call_tool(
|
|
"replace_in_files",
|
|
_ctx(),
|
|
{"find": "Python", "replace": "Rust", "vault": "TestVault"},
|
|
confirm=True,
|
|
)
|
|
assert result.data["dry_run"] is True
|
|
assert result.data["total_matches"] >= 1
|
|
assert (_vault_path() / "note1.md").read_text(encoding="utf-8") == before
|
|
|
|
def test_apply_replaces_and_backs_up(self, client):
|
|
result = call_tool(
|
|
"replace_in_files",
|
|
_ctx(),
|
|
{
|
|
"find": "Python",
|
|
"replace": "Rust",
|
|
"vault": "TestVault",
|
|
"replace_all": True,
|
|
"dry_run": False,
|
|
},
|
|
confirm=True,
|
|
)
|
|
assert result.data["dry_run"] is False
|
|
assert result.data["total_replacements"] >= 1
|
|
assert "Rust" in (_vault_path() / "note1.md").read_text(encoding="utf-8")
|
|
|
|
def test_filters_inaccessible_vaults(self, client):
|
|
ctx = ToolContext(user={"username": "limited", "vaults": ["OtherVault"]}, audit_enabled=False)
|
|
result = call_tool(
|
|
"replace_in_files",
|
|
ctx,
|
|
{"find": "Python", "replace": "Rust", "vault": "all"},
|
|
confirm=True,
|
|
)
|
|
assert result.data["total_matches"] == 0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# D4. Delete / restore
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestDeleteRestore:
|
|
def test_delete_file_with_backup(self, client):
|
|
from backend.services.backups import get_backup_dir
|
|
|
|
result = call_tool("delete_file", _ctx(), {"vault": "TestVault", "path": "note1.md"}, confirm=True)
|
|
assert result.ok
|
|
assert not (_vault_path() / "note1.md").exists()
|
|
assert list(get_backup_dir("TestVault", "note1.md").glob("note1.md.*.bak"))
|
|
|
|
def test_delete_missing_file(self, client):
|
|
with pytest.raises(ToolError) as exc:
|
|
call_tool("delete_file", _ctx(), {"vault": "TestVault", "path": "missing.md"}, confirm=True)
|
|
assert exc.value.code == "not_found"
|
|
|
|
def test_delete_directory(self, client):
|
|
result = call_tool("delete_directory", _ctx(), {"vault": "TestVault", "path": "Projets"}, confirm=True)
|
|
assert result.ok
|
|
assert result.data["deleted_count"] >= 1
|
|
assert not (_vault_path() / "Projets").exists()
|
|
|
|
def test_restore_backup(self, client):
|
|
from backend.services.backups import create_backup
|
|
|
|
file_path = _vault_path() / "note1.md"
|
|
original = file_path.read_text(encoding="utf-8")
|
|
backup_path = create_backup(file_path, "TestVault", "note1.md")
|
|
assert backup_path is not None
|
|
|
|
# Change the file directly (no extra backup), then restore the version.
|
|
file_path.write_text("# modified\n", encoding="utf-8")
|
|
assert file_path.read_text(encoding="utf-8") == "# modified\n"
|
|
|
|
version = int(backup_path.name.split(".")[-2])
|
|
result = call_tool(
|
|
"restore_backup",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "version": version},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
assert file_path.read_text(encoding="utf-8") == original
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Per-vault destructive-tools toggle
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestDestructiveToggle:
|
|
def _disable(self, monkeypatch):
|
|
import backend.vault_settings as vs
|
|
|
|
monkeypatch.setattr(vs, "get_vault_setting", lambda name: {"aiDestructiveTools": False})
|
|
|
|
def test_delete_blocked_when_disabled(self, client, monkeypatch):
|
|
self._disable(monkeypatch)
|
|
with pytest.raises(ToolPermissionError) as exc:
|
|
call_tool("delete_file", _ctx(), {"vault": "TestVault", "path": "note1.md"}, confirm=True)
|
|
assert exc.value.code == "destructive_tools_disabled"
|
|
assert (_vault_path() / "note1.md").exists()
|
|
|
|
def test_rename_blocked_when_disabled(self, client, monkeypatch):
|
|
self._disable(monkeypatch)
|
|
with pytest.raises(ToolPermissionError):
|
|
call_tool(
|
|
"rename_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "new_name": "x.md"},
|
|
confirm=True,
|
|
)
|
|
|
|
def test_edit_still_allowed_when_disabled(self, client, monkeypatch):
|
|
self._disable(monkeypatch)
|
|
result = call_tool(
|
|
"edit_file",
|
|
_ctx(),
|
|
{"vault": "TestVault", "path": "note1.md", "content": "# ok\n"},
|
|
confirm=True,
|
|
)
|
|
assert result.ok
|
|
|
|
def test_replace_filters_disabled_vault(self, client, monkeypatch):
|
|
self._disable(monkeypatch)
|
|
result = call_tool(
|
|
"replace_in_files",
|
|
_ctx(),
|
|
{"find": "Python", "replace": "Rust", "vault": "all", "replace_all": True, "dry_run": False},
|
|
confirm=True,
|
|
)
|
|
assert result.data["total_replacements"] == 0
|