1
1
from __future__ import annotations
2
2
3
3
from collections .abc import AsyncIterator
4
- from typing import Any , Dict , cast
4
+ from typing import Any , cast
5
5
6
6
import pytest
7
7
from openai .types .chat import ChatCompletion , ChatCompletionChunk , ChatCompletionMessage
25
25
26
26
27
27
# Helper functions to create test objects consistently
28
- def create_content_delta (content : str ) -> Dict [str , Any ]:
28
+ def create_content_delta (content : str ) -> dict [str , Any ]:
29
29
"""Create a delta dictionary with regular content"""
30
30
return {
31
31
"content" : content ,
@@ -34,7 +34,7 @@ def create_content_delta(content: str) -> Dict[str, Any]:
34
34
"tool_calls" : None
35
35
}
36
36
37
- def create_reasoning_delta (content : str ) -> Dict [str , Any ]:
37
+ def create_reasoning_delta (content : str ) -> dict [str , Any ]:
38
38
"""Create a delta dictionary with reasoning content. The Only difference is reasoning_content"""
39
39
return {
40
40
"content" : None ,
@@ -45,7 +45,7 @@ def create_reasoning_delta(content: str) -> Dict[str, Any]:
45
45
}
46
46
47
47
48
- def create_chunk (delta : Dict [str , Any ], include_usage : bool = False ) -> ChatCompletionChunk :
48
+ def create_chunk (delta : dict [str , Any ], include_usage : bool = False ) -> ChatCompletionChunk :
49
49
"""Create a ChatCompletionChunk with the given delta"""
50
50
# Create a ChoiceDelta object from the dictionary
51
51
delta_obj = ChoiceDelta (
@@ -54,11 +54,13 @@ def create_chunk(delta: Dict[str, Any], include_usage: bool = False) -> ChatComp
54
54
function_call = delta .get ("function_call" ),
55
55
tool_calls = delta .get ("tool_calls" ),
56
56
)
57
-
57
+
58
58
# Add reasoning_content attribute dynamically if present in the delta
59
59
if "reasoning_content" in delta :
60
- setattr (delta_obj , "reasoning_content" , delta ["reasoning_content" ])
61
-
60
+ # Use direct assignment for the reasoning_content attribute
61
+ delta_obj_any = cast (Any , delta_obj )
62
+ delta_obj_any .reasoning_content = delta ["reasoning_content" ]
63
+
62
64
# Create the chunk
63
65
chunk = ChatCompletionChunk (
64
66
id = "chunk-id" ,
@@ -67,7 +69,7 @@ def create_chunk(delta: Dict[str, Any], include_usage: bool = False) -> ChatComp
67
69
object = "chat.completion.chunk" ,
68
70
choices = [Choice (index = 0 , delta = delta_obj )],
69
71
)
70
-
72
+
71
73
if include_usage :
72
74
chunk .usage = CompletionUsage (
73
75
completion_tokens = 4 ,
@@ -76,7 +78,7 @@ def create_chunk(delta: Dict[str, Any], include_usage: bool = False) -> ChatComp
76
78
completion_tokens_details = CompletionTokensDetails (reasoning_tokens = 2 ),
77
79
prompt_tokens_details = PromptTokensDetails (cached_tokens = 0 ),
78
80
)
79
-
81
+
80
82
return chunk
81
83
82
84
@@ -276,7 +278,7 @@ async def patched_fetch_response(self, *args, **kwargs):
276
278
# verify the final response contains the content
277
279
response_event = output_events [- 1 ]
278
280
assert response_event .type == "response.completed"
279
-
281
+
280
282
# should only have the message, not an empty reasoning item
281
283
assert len (response_event .response .output ) == 1
282
284
assert isinstance (response_event .response .output [0 ], ResponseOutputMessage )
0 commit comments