102 lines
2.7 KiB
Python
102 lines
2.7 KiB
Python
"""
|
|
接口验证: 社区私聊 v1
|
|
路径:
|
|
- GET /api/chat/sessions
|
|
- POST /api/chat/open
|
|
- GET /api/chat/messages
|
|
- POST /api/chat/send
|
|
|
|
运行前请通过环境变量提供两个已登录用户 token：
|
|
PRIVATE_CHAT_TOKEN_A、PRIVATE_CHAT_TOKEN_B。
|
|
"""
|
|
import os
|
|
|
|
import requests
|
|
|
|
|
|
BASE_URL = os.getenv("BASE_URL", "http://127.0.0.1:8000")
|
|
TOKEN_A = os.getenv("PRIVATE_CHAT_TOKEN_A", "")
|
|
TOKEN_B = os.getenv("PRIVATE_CHAT_TOKEN_B", "")
|
|
TARGET_USER_ID = int(os.getenv("PRIVATE_CHAT_TARGET_USER_ID", "0"))
|
|
|
|
|
|
def assert_api_ok(resp):
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["code"] == 1, data
|
|
return data["data"]
|
|
|
|
|
|
def require_env():
|
|
assert TOKEN_A, "请设置 PRIVATE_CHAT_TOKEN_A"
|
|
assert TOKEN_B, "请设置 PRIVATE_CHAT_TOKEN_B"
|
|
assert TARGET_USER_ID > 0, "请设置 PRIVATE_CHAT_TARGET_USER_ID"
|
|
|
|
|
|
def test_private_chat_flow():
|
|
require_env()
|
|
|
|
headers_a = {"token": TOKEN_A}
|
|
headers_b = {"token": TOKEN_B}
|
|
|
|
session = assert_api_ok(
|
|
requests.post(
|
|
f"{BASE_URL}/api/chat/open",
|
|
json={"target_user_id": TARGET_USER_ID},
|
|
headers=headers_a,
|
|
timeout=10,
|
|
)
|
|
)
|
|
assert session["target_user"]["id"] == TARGET_USER_ID
|
|
assert "is_mutual" in session["relationship"]
|
|
assert "relationship_text" in session["relationship"]
|
|
|
|
message = assert_api_ok(
|
|
requests.post(
|
|
f"{BASE_URL}/api/chat/send",
|
|
json={
|
|
"session_id": session["id"],
|
|
"message_type": "text",
|
|
"content": "私聊接口冒烟测试",
|
|
},
|
|
headers=headers_a,
|
|
timeout=10,
|
|
)
|
|
)
|
|
assert message["message_type"] == "text"
|
|
assert message["content"] == "私聊接口冒烟测试"
|
|
|
|
messages = assert_api_ok(
|
|
requests.get(
|
|
f"{BASE_URL}/api/chat/messages",
|
|
params={"session_id": session["id"], "after_id": 0},
|
|
headers=headers_b,
|
|
timeout=10,
|
|
)
|
|
)
|
|
assert any(item["id"] == message["id"] for item in messages["lists"])
|
|
|
|
image_message = assert_api_ok(
|
|
requests.post(
|
|
f"{BASE_URL}/api/chat/send",
|
|
json={
|
|
"session_id": session["id"],
|
|
"message_type": "image",
|
|
"content": "/uploads/private-chat-test.png",
|
|
},
|
|
headers=headers_a,
|
|
timeout=10,
|
|
)
|
|
)
|
|
assert image_message["message_type"] == "image"
|
|
|
|
sessions = assert_api_ok(
|
|
requests.get(f"{BASE_URL}/api/chat/sessions", headers=headers_a, timeout=10)
|
|
)
|
|
assert any(item["id"] == session["id"] for item in sessions["lists"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_private_chat_flow()
|
|
print("私聊接口冒烟通过")
|