no message

qa/backend/test_private_chat_api.py

@@ -0,0 +1,101 @@
+"""
+接口验证: 社区私聊 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("私聊接口冒烟通过")