ObsiGate/tests/test_tools_mutations.py

# 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()


# ═══════════════════════════════════════════════════════════════════
# 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