beaver-agent/tests/test_policy.py

import tempfile
from pathlib import Path

import pytest
from beaver_gateway.core.policy import Deny, ToolCall

from policy import Zones, bash_zones, requires_skill, skill_tracker, vault_zones


@pytest.fixture
def vault() -> Path:
    root = Path(tempfile.mkdtemp(prefix="beaver-vault-"))
    for d in ("мета/бобер", "💬 чаты", "📅 дни", "👤 люди"):
        (root / d).mkdir(parents=True)
    (root / "💬 чаты/старый.md").write_text("x")
    (root / "📅 дни/2026-08-29.md").write_text("x")
    return root


@pytest.fixture
def zones(vault: Path) -> Zones:
    return Zones(
        vault=vault, write=(vault / "мета/бобер",), create=(vault / "💬 чаты",)
    )


def call(vault: Path, tool: str, state=None, **tool_input) -> ToolCall:
    return ToolCall(
        tool=tool,
        input=tool_input,
        agent="a",
        kind="master",
        conversation="c",
        cwd=vault,
        state=state if state is not None else {},
    )


def test_write_outside_zones_denied(vault, zones):
    rule = vault_zones(zones)
    deny = rule(call(vault, "Write", file_path=str(vault / "📅 дни/2026-08-29.md")))
    assert isinstance(deny, Deny) and "мета/бобер" in deny.reason
    assert rule(call(vault, "Edit", file_path=str(vault / "👤 люди/x.md"))) is not None
    assert rule(call(vault, "Write", file_path="новое.md")) is not None


def test_zones_allow_and_create_only(vault, zones):
    rule = vault_zones(zones)
    assert (
        rule(call(vault, "Write", file_path=str(vault / "мета/бобер/дни/x.md"))) is None
    )
    assert rule(call(vault, "Edit", file_path="мета/бобер/состояние.md")) is None
    assert rule(call(vault, "Write", file_path=str(vault / "💬 чаты/новый.md"))) is None
    assert (
        rule(call(vault, "Write", file_path=str(vault / "💬 чаты/старый.md")))
        is not None
    )
    assert (
        rule(call(vault, "Edit", file_path=str(vault / "💬 чаты/новый.md"))) is not None
    )
    assert rule(call(vault, "Read", file_path=str(vault / "📅 дни/x.md"))) is None
    assert rule(call(vault, "Write", file_path="/tmp/scratch.md")) is None


@pytest.mark.parametrize(
    "command",
    [
        "rm '📅 дни/2026-08-29.md'",
        "rm -rf 👤\\ люди",
        "mv '👤 люди/x.md' 'мета/бобер/x.md'",
        "cp мета/бобер/x.md '📅 дни/y.md'",
        "echo hi > '📅 дни/today.md'",
        "cat a.md | tee '💬 чаты/старый.md'",
        "sed -i 's/a/b/' '👤 люди/x.md'",
        "cd 👤\\ люди && rm x.md",
        "ls; touch new.md",
    ],
)
def test_bash_mutations_outside_zones_denied(vault, zones, command):
    deny = bash_zones(zones)(call(vault, "Bash", command=command))
    assert isinstance(deny, Deny), command
    assert deny.reason.startswith("Bash: ")


@pytest.mark.parametrize(
    "command",
    [
        "cat '👤 люди/x.md' | grep foo",
        "rm мета/бобер/дни/old.md",
        "echo hi > мета/бобер/x.md 2>&1",
        "cp '👤 люди/x.md' мета/бобер/копия.md",
        "sed 's/a/b/' '👤 люди/x.md'",
        "echo hi > '💬 чаты/новый.md'",
        "rm /tmp/x && ls > /dev/null",
        "python3 -c \"print('x')\" >> /tmp/log",
        "grep -r 'x' . --include='*.md'",
    ],
)
def test_bash_reads_and_zone_writes_allowed(vault, zones, command):
    assert bash_zones(zones)(call(vault, "Bash", command=command)) is None, command


def test_bash_unbalanced_quotes_fall_back(vault, zones):
    assert bash_zones(zones)(call(vault, "Bash", command='rm "📅')) is not None


def test_firefly_requires_open_skill(vault):
    state = {}
    tracker, gate = (
        skill_tracker(),
        requires_skill("firefly", ("mcp__firefly__store_*",)),
    )
    store = call(vault, "mcp__firefly__store_transaction", state=state, data={})
    assert isinstance(gate(store), Deny)
    assert gate(call(vault, "mcp__firefly__list_account", state=state)) is None
    tracker(call(vault, "Skill", state=state, skill="vault:firefly"))
    assert gate(store) is None


@pytest.fixture
def home(vault: Path) -> Zones:
    """Зоны диспетчера с 2026-08-30: vault - дом, удаление - только своё."""
    (vault / ".obsidian").mkdir()
    (vault / "мета/шаблоны").mkdir()
    return Zones(
        vault=vault,
        write=(vault / "мета/бобер",),
        create=(vault / "💬 чаты",),
        edit=(vault,),
        protected=(vault / ".obsidian", vault / "мета"),
    )


def test_home_zones_allow_notes_but_protect_obsidian_and_meta(vault, home):
    rule = vault_zones(home)
    assert rule(call(vault, "Write", file_path=str(vault / "👤 люди/новый.md"))) is None
    assert (
        rule(call(vault, "Edit", file_path=str(vault / "📅 дни/2026-08-29.md"))) is None
    )
    assert rule(call(vault, "Write", file_path=str(vault / "мета/бобер/x.md"))) is None
    assert (
        rule(call(vault, "Edit", file_path=str(vault / "💬 чаты/старый.md")))
        is not None
    )
    deny = rule(call(vault, "Write", file_path=str(vault / ".obsidian/app.json")))
    assert isinstance(deny, Deny) and "Бобра" in deny.reason
    assert (
        rule(call(vault, "Edit", file_path=str(vault / "мета/шаблоны/x.md")))
        is not None
    )


@pytest.mark.parametrize(
    "command",
    [
        "mv '👤 люди/x.md' '👤 люди/архив/x.md'",
        "echo hi >> '📅 дни/2026-08-29.md'",
        "sed -i 's/a/b/' '👤 люди/x.md'",
        "rm мета/бобер/дни/old.md",
        "mkdir '💻 проекты/новый'",
    ],
)
def test_home_bash_edits_allowed(vault, home, command):
    assert bash_zones(home)(call(vault, "Bash", command=command)) is None, command


@pytest.mark.parametrize(
    "command",
    [
        "rm '📅 дни/2026-08-29.md'",
        "rm -rf 👤\\ люди",
        "truncate -s 0 '👤 люди/x.md'",
        "echo x > .obsidian/app.json",
        "mv '👤 люди/x.md' мета/шаблоны/x.md",
    ],
)
def test_home_bash_deletes_and_protected_denied(vault, home, command):
    deny = bash_zones(home)(call(vault, "Bash", command=command))
    assert isinstance(deny, Deny), command