94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
|
|
|||
|
|
import json
|
|||
|
|
import httpx
|
|||
|
|
import os
|
|||
|
|
|
|||
|
|
# --- 配置 ---
|
|||
|
|
SERVER_URL = "http://localhost:8000"
|
|||
|
|
GRAPH_EXECUTE_ENDPOINT = f"{SERVER_URL}/execute_graph"
|
|||
|
|
GRAPH_FILE_PATH = os.path.join(os.path.dirname(__file__), "assets", "complex_remote_graph.json")
|
|||
|
|
|
|||
|
|
def run_test():
|
|||
|
|
"""
|
|||
|
|
执行端到端远程调用测试。
|
|||
|
|
|
|||
|
|
使用方法:
|
|||
|
|
1. 确保 remote_test_server.py 正在运行 (uvicorn server.remote_test_server:app --reload --port 8000)
|
|||
|
|
2. 确保 test_complex_remote_client.py 正在运行 (python tests/test_complex_remote_client.py)
|
|||
|
|
3. 运行此脚本 (python tests/run_remote_test.py)
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
print("--- Running End-to-End Remote Execution Test ---")
|
|||
|
|
|
|||
|
|
# 1. 加载图JSON文件
|
|||
|
|
try:
|
|||
|
|
with open(GRAPH_FILE_PATH, 'r', encoding='utf-8') as f:
|
|||
|
|
graph_data = json.load(f)
|
|||
|
|
print(f"Successfully loaded graph from: {GRAPH_FILE_PATH}")
|
|||
|
|
except FileNotFoundError:
|
|||
|
|
print(f"ERROR: Graph file not found at {GRAPH_FILE_PATH}")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 2. 准备初始输入
|
|||
|
|
# 根据 complex_remote_graph.json 的 "interface" 定义
|
|||
|
|
initial_inputs = {
|
|||
|
|
"main_text": "hello_remote_world",
|
|||
|
|
"main_num": 5
|
|||
|
|
}
|
|||
|
|
print(f"Initial inputs: {initial_inputs}")
|
|||
|
|
|
|||
|
|
# 3. 构建请求体
|
|||
|
|
request_payload = {
|
|||
|
|
"graph_data": graph_data,
|
|||
|
|
"initial_inputs": initial_inputs
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 4. 发送HTTP请求到服务器
|
|||
|
|
print(f"Sending POST request to {GRAPH_EXECUTE_ENDPOINT}...")
|
|||
|
|
try:
|
|||
|
|
with httpx.Client(timeout=60.0) as client:
|
|||
|
|
response = client.post(GRAPH_EXECUTE_ENDPOINT, json=request_payload)
|
|||
|
|
response.raise_for_status() # 如果状态码不是 2xx，则会引发异常
|
|||
|
|
|
|||
|
|
result = response.json()
|
|||
|
|
|
|||
|
|
print("\n--- Test Result ---")
|
|||
|
|
print("Request successful!")
|
|||
|
|
print("Server response:")
|
|||
|
|
# 使用json.dumps美化输出
|
|||
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|||
|
|
|
|||
|
|
# 预期结果计算：
|
|||
|
|
# 1. main_text "hello_remote_world" 长度为 18
|
|||
|
|
# 2. 子图接收 sub_in_1=18, sub_in_2=5
|
|||
|
|
# 3. 子图内本地相加: 18 + 5 = 23
|
|||
|
|
# 4. 子图内远程计算比率: 23 / 10.0 = 2.3
|
|||
|
|
# 5. 主图接收到最终结果 2.3
|
|||
|
|
expected_ratio = 2.3
|
|||
|
|
final_ratio = result.get("outputs", {}).get("final_ratio", {}).get("value")
|
|||
|
|
|
|||
|
|
print("\n--- Verification ---")
|
|||
|
|
print(f"Expected final_ratio: {expected_ratio}")
|
|||
|
|
print(f"Actual final_ratio: {final_ratio}")
|
|||
|
|
assert final_ratio == expected_ratio, "Test assertion failed!"
|
|||
|
|
print("✅ Test Passed Successfully!")
|
|||
|
|
|
|||
|
|
|
|||
|
|
except httpx.RequestError as e:
|
|||
|
|
print("\n--- TEST FAILED ---")
|
|||
|
|
print(f"ERROR: Could not connect to the server at {SERVER_URL}.")
|
|||
|
|
print("Please ensure the 'remote_test_server.py' is running.")
|
|||
|
|
print(f"Details: {e}")
|
|||
|
|
except httpx.HTTPStatusError as e:
|
|||
|
|
print("\n--- TEST FAILED ---")
|
|||
|
|
print(f"ERROR: Server returned a non-200 status code: {e.response.status_code}")
|
|||
|
|
print("Server response:")
|
|||
|
|
try:
|
|||
|
|
# 尝试打印JSON错误详情
|
|||
|
|
print(json.dumps(e.response.json(), indent=2, ensure_ascii=False))
|
|||
|
|
except json.JSONDecodeError:
|
|||
|
|
print(e.response.text)
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
run_test()
|