forked from openai/openai-agents-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_run_context_wrapper.py
More file actions
122 lines (83 loc) · 4.56 KB
/
Copy pathtest_run_context_wrapper.py
File metadata and controls
122 lines (83 loc) · 4.56 KB
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
from typing import Any
from agents.items import ToolApprovalItem
from agents.run_context import RunContextWrapper
from tests.utils.hitl import make_agent
class BrokenStr:
def __str__(self) -> str:
raise RuntimeError("broken")
def test_run_context_to_str_or_none_handles_errors() -> None:
assert RunContextWrapper._to_str_or_none("ok") == "ok"
assert RunContextWrapper._to_str_or_none(123) == "123"
assert RunContextWrapper._to_str_or_none(BrokenStr()) is None
assert RunContextWrapper._to_str_or_none(None) is None
def test_run_context_resolve_tool_name_and_call_id_fallbacks() -> None:
raw: dict[str, Any] = {"name": "raw_tool", "id": "raw-id"}
item = ToolApprovalItem(agent=make_agent(), raw_item=raw, tool_name=None)
assert RunContextWrapper._resolve_tool_name(item) == "raw_tool"
assert RunContextWrapper._resolve_call_id(item) == "raw-id"
def test_run_context_scopes_approvals_to_call_ids() -> None:
wrapper: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
agent = make_agent()
approval = ToolApprovalItem(agent=agent, raw_item={"type": "tool_call", "call_id": "call-1"})
wrapper.approve_tool(approval)
assert wrapper.is_tool_approved("tool_call", "call-1") is True
# A different call ID should require a fresh approval.
assert wrapper.is_tool_approved("tool_call", "call-2") is None
def test_run_context_scopes_rejections_to_call_ids() -> None:
wrapper: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
agent = make_agent()
approval = ToolApprovalItem(agent=agent, raw_item={"type": "tool_call", "call_id": "call-1"})
wrapper.reject_tool(approval)
assert wrapper.is_tool_approved("tool_call", "call-1") is False
# A different call ID should require a fresh approval.
assert wrapper.is_tool_approved("tool_call", "call-2") is None
def test_run_context_honors_global_approval_and_rejection() -> None:
wrapper: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
agent = make_agent()
approval = ToolApprovalItem(agent=agent, raw_item={"type": "tool_call", "call_id": "call-1"})
wrapper.approve_tool(approval, always_approve=True)
assert wrapper.is_tool_approved("tool_call", "call-2") is True
wrapper.reject_tool(approval, always_reject=True)
assert wrapper.is_tool_approved("tool_call", "call-3") is False
def test_run_context_stores_per_call_rejection_messages() -> None:
wrapper: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
agent = make_agent()
approval = ToolApprovalItem(agent=agent, raw_item={"type": "tool_call", "call_id": "call-1"})
wrapper.reject_tool(approval, rejection_message="Denied by policy")
assert wrapper.get_rejection_message("tool_call", "call-1") == "Denied by policy"
assert wrapper.get_rejection_message("tool_call", "call-2") is None
def test_run_context_stores_sticky_rejection_messages_for_always_reject() -> None:
wrapper: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
agent = make_agent()
approval = ToolApprovalItem(agent=agent, raw_item={"type": "tool_call", "call_id": "call-1"})
wrapper.reject_tool(approval, always_reject=True, rejection_message="")
assert wrapper.get_rejection_message("tool_call", "call-1") == ""
assert wrapper.get_rejection_message("tool_call", "call-2") == ""
def test_run_context_clears_rejection_message_after_approval() -> None:
wrapper: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
agent = make_agent()
approval = ToolApprovalItem(agent=agent, raw_item={"type": "tool_call", "call_id": "call-1"})
wrapper.reject_tool(approval, rejection_message="Denied by policy")
wrapper.approve_tool(approval)
assert wrapper.get_rejection_message("tool_call", "call-1") is None
def test_run_context_unknown_tool_name_fallback() -> None:
agent = make_agent()
raw: dict[str, Any] = {}
approval = ToolApprovalItem(agent=agent, raw_item=raw, tool_name=None)
assert RunContextWrapper._resolve_tool_name(approval) == "unknown_tool"
def test_tool_approval_item_preserves_positional_type_argument() -> None:
raw: dict[str, Any] = {
"type": "function_call",
"name": "lookup_account",
"call_id": "call-1",
"namespace": "billing",
}
approval = ToolApprovalItem(
make_agent(),
raw,
"lookup_account",
"tool_approval_item",
)
assert approval.type == "tool_approval_item"
assert approval.tool_name == "lookup_account"
assert approval.tool_namespace == "billing"