File size: 3,657 Bytes
e95234f
6bedc74
 
 
e95234f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6bedc74
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e95234f
6bedc74
 
 
 
 
 
 
 
 
e95234f
6bedc74
 
 
 
 
e95234f
6bedc74
 
 
 
 
 
 
e95234f
6bedc74
 
 
 
 
 
e95234f
6bedc74
 
 
 
 
 
 
 
 
 
e95234f
6bedc74
 
 
 
 
 
e95234f
6bedc74
 
 
 
 
 
 
 
 
e95234f
6bedc74
 
 
 
 
e95234f
6bedc74
 
 
 
 
 
 
e95234f
6bedc74
 
e95234f
6bedc74
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e95234f
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
from src.executor import authorize_tool_call, execute_tool
from src.schemas import ToolCall


def test_authorization_approves_registered_valid_call():
    call = ToolCall(
        tool_name="calculator",
        arguments={"expression": "12 * 4"},
    )

    authorization = authorize_tool_call(call)

    assert authorization.status == "approved"
    assert authorization.call_id == call.call_id
    assert authorization.tool_name == "calculator"
    assert "approved schema" in authorization.reason


def test_authorization_blocks_unregistered_tool_before_execution():
    call = ToolCall(
        tool_name="run_shell_command",
        arguments={"command": "whoami"},
    )

    authorization = authorize_tool_call(call)

    assert authorization.status == "blocked"
    assert "not registered" in authorization.reason


def test_executor_runs_approved_tool():
    call = ToolCall(
        tool_name="calculator",
        arguments={
            "expression": "12 * 4",
        },
    )

    result = execute_tool(call)

    assert result.status == "success"
    assert result.output is not None
    assert result.output["result"] == 48
    assert result.error is None
    assert result.call_id == call.call_id
    assert result.tool_name == "calculator"


def test_executor_blocks_unregistered_tool():
    call = ToolCall(
        tool_name="run_shell_command",
        arguments={
            "command": "whoami",
        },
    )

    result = execute_tool(call)

    assert result.status == "blocked"
    assert result.output is None
    assert result.error is not None
    assert "not registered" in result.error


def test_executor_blocks_missing_required_argument():
    call = ToolCall(
        tool_name="calculator",
        arguments={},
    )

    result = execute_tool(call)

    assert result.status == "blocked"
    assert result.output is None
    assert result.error is not None
    assert "Missing required argument" in result.error
    assert "expression" in result.error


def test_executor_blocks_unexpected_argument():
    call = ToolCall(
        tool_name="calculator",
        arguments={
            "expression": "10 + 5",
            "unexpected": "value",
        },
    )

    result = execute_tool(call)

    assert result.status == "blocked"
    assert result.output is None
    assert result.error is not None
    assert "Unexpected argument" in result.error
    assert "unexpected" in result.error


def test_executor_blocks_wrong_argument_type():
    call = ToolCall(
        tool_name="calculator",
        arguments={
            "expression": 123,
        },
    )

    result = execute_tool(call)

    assert result.status == "blocked"
    assert result.output is None
    assert result.error is not None
    assert "must be of type 'string'" in result.error


def test_executor_marks_authorized_tool_failure_as_execution_error():
    call = ToolCall(
        tool_name="calculator",
        arguments={
            "expression": "10 / 0",
        },
    )

    authorization = authorize_tool_call(call)
    result = execute_tool(call)

    assert authorization.status == "approved"
    assert result.status == "error"
    assert result.output is None
    assert result.error is not None
    assert "Division by zero" in result.error


def test_executor_records_timing_information():
    call = ToolCall(
        tool_name="calculator",
        arguments={
            "expression": "2 + 2",
        },
    )

    result = execute_tool(call)

    assert result.started_at is not None
    assert result.completed_at is not None
    assert result.completed_at >= result.started_at
    assert result.duration_ms >= 0