104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
测试 @Agent 功能
|
||
|
|
"""
|
||
|
|
|
||
|
|
import requests
|
||
|
|
|
||
|
|
API_BASE = 'http://localhost:8000/api/v1'
|
||
|
|
|
||
|
|
def test_mention_agent():
|
||
|
|
print("="*60)
|
||
|
|
print("📍 测试 @Agent 功能")
|
||
|
|
print("="*60)
|
||
|
|
|
||
|
|
# 登录
|
||
|
|
res = requests.post(f'{API_BASE}/auth/login/', json={
|
||
|
|
'username': 'test',
|
||
|
|
'password': 'test123'
|
||
|
|
})
|
||
|
|
token = res.json()['token']
|
||
|
|
headers = {'Authorization': f'Bearer {token}'}
|
||
|
|
print(f"✅ 登录成功")
|
||
|
|
|
||
|
|
# 创建会议
|
||
|
|
res = requests.post(f'{API_BASE}/meetings/', json={
|
||
|
|
'topic': '@Agent 测试会议'
|
||
|
|
}, headers=headers)
|
||
|
|
meeting_id = res.json()['id']
|
||
|
|
print(f"✅ 创建会议：{meeting_id}")
|
||
|
|
|
||
|
|
# Agent A 加入（查信箱自动加入）
|
||
|
|
res = requests.get(
|
||
|
|
f'{API_BASE}/meetings/{meeting_id}/inbox/',
|
||
|
|
params={'agent_id': 'agent_a', 'agent_name': '助手 A', 'agent_emoji': '🤖'}
|
||
|
|
)
|
||
|
|
print(f"✅ Agent A 加入会议")
|
||
|
|
|
||
|
|
# Agent B 加入
|
||
|
|
res = requests.get(
|
||
|
|
f'{API_BASE}/meetings/{meeting_id}/inbox/',
|
||
|
|
params={'agent_id': 'agent_b', 'agent_name': '助手 B', 'agent_emoji': '🦊'}
|
||
|
|
)
|
||
|
|
print(f"✅ Agent B 加入会议")
|
||
|
|
|
||
|
|
# 获取参会者列表
|
||
|
|
res = requests.get(f'{API_BASE}/meetings/{meeting_id}/participants/')
|
||
|
|
participants = res.json()
|
||
|
|
print(f"✅ 参会者列表：{len(participants)} 人")
|
||
|
|
for p in participants:
|
||
|
|
print(f" - {p['nickname']} (ID: {p['agent_id']})")
|
||
|
|
|
||
|
|
# @Agent A
|
||
|
|
res = requests.post(f'{API_BASE}/meetings/{meeting_id}/mention_agent/', json={
|
||
|
|
'target_agent_id': 'agent_a',
|
||
|
|
'content': '请汇报一下进度',
|
||
|
|
'sender_agent_id': 'human_user',
|
||
|
|
'sender_name': '北极星',
|
||
|
|
'sender_emoji': '⭐'
|
||
|
|
})
|
||
|
|
if res.status_code == 201:
|
||
|
|
msg = res.json()
|
||
|
|
print(f"✅ @Agent A 成功：{msg['content']}")
|
||
|
|
else:
|
||
|
|
print(f"❌ @Agent 失败：{res.text}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Agent A 查信箱（应该收到 @ 消息）
|
||
|
|
res = requests.get(
|
||
|
|
f'{API_BASE}/meetings/{meeting_id}/inbox/',
|
||
|
|
params={'agent_id': 'agent_a', 'agent_name': '助手 A', 'agent_emoji': '🤖'}
|
||
|
|
)
|
||
|
|
inbox = res.json()
|
||
|
|
print(f"✅ Agent A 信箱：{inbox['unread_count']} 条未读")
|
||
|
|
|
||
|
|
# Agent A 回复
|
||
|
|
res = requests.post(f'{API_BASE}/meetings/{meeting_id}/agent_reply/', json={
|
||
|
|
'agent_id': 'agent_a',
|
||
|
|
'agent_name': '助手 A',
|
||
|
|
'agent_emoji': '🤖',
|
||
|
|
'content': '收到！进度正常，已完成 80%。',
|
||
|
|
'in_reply_to': msg['id']
|
||
|
|
})
|
||
|
|
if res.status_code == 201:
|
||
|
|
print(f"✅ Agent A 回复成功")
|
||
|
|
else:
|
||
|
|
print(f"❌ Agent A 回复失败：{res.text}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
# 获取全部消息
|
||
|
|
res = requests.get(f'{API_BASE}/meetings/{meeting_id}/messages/?last_id=0')
|
||
|
|
messages = res.json()['messages']
|
||
|
|
print(f"\n💬 全部消息 ({len(messages)} 条):")
|
||
|
|
for m in messages:
|
||
|
|
reply_info = f" (回复 #{m['in_reply_to']})" if m.get('in_reply_to') else ""
|
||
|
|
print(f" {m['sender_emoji']} {m['sender_name']}: {m['content']}{reply_info}")
|
||
|
|
|
||
|
|
print("\n" + "="*60)
|
||
|
|
print("✅ @Agent 功能测试通过！")
|
||
|
|
print("="*60)
|
||
|
|
return True
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
test_mention_agent()
|