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
147
148
149
150
151
152
|
# pylint: disable=too-many-lines,line-too-long,useless-suppression
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
# cSpell:disable
import pytest
from test_base import TestBase, servicePreparer
from devtools_testutils import is_live_and_not_recording
from azure.ai.projects.models import (
# ResponsesUserMessageItemParam,
# ResponsesSystemMessageItemParam,
ItemContentInputText,
ItemType,
ResponsesMessageRole,
ItemContentType,
)
# TODO: Emitter did not produce the output class OpenAI.ConversationItemList. Validating service response as Dict for now.
class TestConversationItemsCrud(TestBase):
@servicePreparer()
@pytest.mark.skipif(
condition=(not is_live_and_not_recording()),
reason="Skipped because we cannot record network calls with OpenAI client",
)
def test_conversation_items_crud(self, **kwargs):
"""
Test CRUD operations for Conversation Items.
This test gets an OpenAI client, creates a conversation, then performs CRUD
operations on items within it:
- create_items: Add items to the conversation
- list_items: List all items in the conversation
- delete_item: Delete specific items from the conversation
It uses different ways of creating items: strongly typed or dictionary.
Routes used in this test:
Action REST API Route OpenAI Client Method
------+-------------------------------------------------------+-----------------------------------
POST /openai/conversations client.conversations.create()
POST /openai/conversations/{conversation_id}/items client.conversations.items.create()
GET /openai/conversations/{conversation_id}/items/{item_id} client.conversations.items.retrieve()
GET /openai/conversations/{conversation_id}/items client.conversations.items.list()
DELETE /openai/conversations/{conversation_id}/items/{item_id} client.conversations.items.delete()
DELETE /openai/conversations/{conversation_id} client.conversations.delete()
"""
client = self.create_client(operation_group="agents", **kwargs).get_openai_client()
# Create a conversation to work with
conversation = client.conversations.create()
print(f"Created conversation (id: {conversation.id})")
try:
print(f"Test create_items")
# Create items with short-form and long-form text message as Dict
# See https://platform.openai.com/docs/api-reference/conversations/create-items
items = [
{"type": "message", "role": "user", "content": "first message"},
{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "second message"}]},
]
items = client.conversations.items.create(
conversation.id,
items=items,
)
assert items.has_more is False
item_list = items.data
print(f"Created item with short-form and long form text messages as Dict")
assert len(item_list) == 2
self._validate_conversation_item(
item_list[0],
expected_type=ItemType.MESSAGE,
expected_role=ResponsesMessageRole.USER,
expected_content_type=ItemContentType.INPUT_TEXT,
expected_content_text="first message",
)
self._validate_conversation_item(
item_list[1],
expected_type=ItemType.MESSAGE,
expected_role=ResponsesMessageRole.USER,
expected_content_type=ItemContentType.INPUT_TEXT,
expected_content_text="second message",
)
item1_id = item_list[0].id
item2_id = item_list[1].id
# Create 2 items, one system message with short-form strongly typed, and user message with long-form strongly typed
# items = client.conversations.items.create(
# conversation.id,
# items=[
# ResponsesSystemMessageItemParam(content="third message"),
# ResponsesUserMessageItemParam(content=[ItemContentInputText(text="fourth message")]),
# ],
# )
# assert items.has_more is False
# item_list = items.data
# print(f"Created 2 strongly typed items")
# assert len(item_list) == 2
# self._validate_conversation_item(
# item_list[0],
# expected_type=ItemType.MESSAGE,
# expected_role=ResponsesMessageRole.SYSTEM,
# expected_content_type=ItemContentType.INPUT_TEXT,
# expected_content_text="third message",
# )
# self._validate_conversation_item(
# item_list[1],
# expected_type=ItemType.MESSAGE,
# expected_role=ResponsesMessageRole.USER,
# expected_content_type=ItemContentType.INPUT_TEXT,
# expected_content_text="fourth message",
# )
# item3_id = item_list[0].id
# item4_id = item_list[1].id
print(f"Test retrieve item")
item = client.conversations.items.retrieve(conversation_id=conversation.id, item_id=item1_id)
self._validate_conversation_item(
item,
expected_type=ItemType.MESSAGE,
expected_id=item1_id,
expected_role=ResponsesMessageRole.USER,
expected_content_type=ItemContentType.INPUT_TEXT,
expected_content_text="first message",
)
print(f"Test list items")
item_count = 0
for item in client.conversations.items.list(conversation.id):
item_count += 1
self._validate_conversation_item(item)
assert item_count == 2
print(f"Test delete item")
# result = client.conversations.items.delete(conversation_id=conversation.id, item_id=item4_id)
# assert result.id == conversation.id
result = client.conversations.items.delete(conversation_id=conversation.id, item_id=item2_id)
assert result.id == conversation.id
# Verify items were deleted by listing again
remaining_items = list(client.conversations.items.list(conversation.id))
assert len(remaining_items) == 1
finally:
# Clean up the conversation
conversation_result = client.conversations.delete(conversation.id)
print(f"Conversation deleted (id: {conversation_result.id}, deleted: {conversation_result.deleted})")
|
