from pathlib import Path import pytest from beaver_gateway.agents.policy import Deny, ToolCall, brief, evaluate, hook_output def call(tool: str, **tool_input) -> ToolCall: return ToolCall( tool=tool, input=tool_input, agent="a", kind="master", conversation="c1", cwd=Path("/vault"), ) def test_path_resolves_relative_against_cwd(): assert call("Write", file_path="мета/бобер/x.md").path() == Path( "/vault/мета/бобер/x.md" ) assert call("Write", file_path="/vault/a/../b.md").path() == Path("/vault/b.md") assert call("Write", file_path="../../etc/passwd").path() == Path("/etc/passwd") assert call("Bash", command="ls").path() is None def test_mcp_split(): assert call("mcp__firefly__store_transaction").mcp == ( "firefly", "store_transaction", ) assert call("Write").mcp is None async def test_first_deny_wins_and_async_rules_work(): seen = [] def allow(c): seen.append("allow") async def deny(c): return Deny(reason="no") def never(c): raise AssertionError verdict = await evaluate((allow, deny, never), call("Write")) assert verdict == Deny(reason="no") assert seen == ["allow"] async def test_raising_rule_fails_closed(): def boom(c): raise RuntimeError("x") verdict = await evaluate((boom,), call("Write")) assert verdict is not None and "boom" in verdict.reason def test_hook_output_shapes(): assert hook_output(None) == {} out = hook_output(Deny(reason="r")) assert out["hookSpecificOutput"]["permissionDecision"] == "deny" assert out["hookSpecificOutput"]["permissionDecisionReason"] == "r" def test_brief_prefers_path_then_command_and_truncates(): assert brief({"file_path": "/v/x.md", "content": "a" * 999}) == "/v/x.md" assert brief({"command": "ls -la"}) == "ls -la" assert len(brief({"q": "a" * 999})) == 240