172 lines
4.7 KiB
Python
172 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
简单的文件上传测试脚本
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
|
||
# API基础URL
|
||
BASE_URL = "http://localhost:8000/api/v1"
|
||
|
||
def test_auth():
|
||
"""测试认证"""
|
||
# 先注册
|
||
register_data = {
|
||
"username": "testuser",
|
||
"email": "test@example.com",
|
||
"password": "TestPass123!",
|
||
"confirm_password": "TestPass123!"
|
||
}
|
||
|
||
try:
|
||
print("尝试注册用户...")
|
||
reg_response = requests.post(f"{BASE_URL}/auth/register", json=register_data)
|
||
print(f"注册状态码: {reg_response.status_code}")
|
||
|
||
if reg_response.status_code == 201:
|
||
reg_result = reg_response.json()
|
||
if reg_result.get("success"):
|
||
token = reg_result["data"]["tokens"]["access_token"]
|
||
print("注册成功，获得token")
|
||
return token
|
||
else:
|
||
print(f"注册失败: {reg_result}")
|
||
else:
|
||
print(f"注册请求失败，状态码: {reg_response.status_code}")
|
||
|
||
# 如果注册失败，尝试登录
|
||
print("尝试登录用户...")
|
||
login_data = {
|
||
"username": "testuser",
|
||
"password": "testpass123"
|
||
}
|
||
|
||
response = requests.post(f"{BASE_URL}/auth/login", json=login_data)
|
||
print(f"登录状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
print(f"登录结果: {result.get('success', False)}")
|
||
if result.get("success"):
|
||
token = result["data"]["tokens"]["access_token"]
|
||
print("登录成功，获得token")
|
||
return token
|
||
else:
|
||
print(f"登录失败: {result}")
|
||
|
||
return None
|
||
except Exception as e:
|
||
print(f"认证出错: {e}")
|
||
return None
|
||
|
||
def test_file_upload(token):
|
||
"""测试文件上传"""
|
||
headers = {
|
||
"Authorization": f"Bearer {token}"
|
||
}
|
||
|
||
# 创建测试文件
|
||
test_content = "Hello World! 测试文件内容"
|
||
|
||
try:
|
||
files = {
|
||
"file": ("test.txt", test_content.encode("utf-8"), "text/plain")
|
||
}
|
||
data = {
|
||
"description": "测试文件",
|
||
"tags": "test,upload",
|
||
"is_public": "false"
|
||
}
|
||
|
||
response = requests.post(
|
||
f"{BASE_URL}/files/upload",
|
||
headers=headers,
|
||
files=files,
|
||
data=data
|
||
)
|
||
|
||
print(f"上传状态码: {response.status_code}")
|
||
print(f"上传响应: {response.text}")
|
||
|
||
if response.status_code == 201:
|
||
result = response.json()
|
||
if result.get("success"):
|
||
file_info = result["data"]["file"]
|
||
print(f"上传成功! 文件ID: {file_info['id']}")
|
||
return file_info['id']
|
||
|
||
return None
|
||
except Exception as e:
|
||
print(f"上传出错: {e}")
|
||
return None
|
||
|
||
def test_file_list(token):
|
||
"""测试获取文件列表"""
|
||
headers = {
|
||
"Authorization": f"Bearer {token}"
|
||
}
|
||
|
||
try:
|
||
response = requests.get(
|
||
f"{BASE_URL}/files/list",
|
||
headers=headers
|
||
)
|
||
|
||
print(f"列表状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("success"):
|
||
files = result["data"]["files"]
|
||
print(f"文件数量: {len(files)}")
|
||
for f in files:
|
||
print(f" - {f['original_filename']} ({f['file_size']} bytes)")
|
||
|
||
except Exception as e:
|
||
print(f"获取列表出错: {e}")
|
||
|
||
def test_storage_info(token):
|
||
"""测试存储信息"""
|
||
headers = {
|
||
"Authorization": f"Bearer {token}"
|
||
}
|
||
|
||
try:
|
||
response = requests.get(
|
||
f"{BASE_URL}/files/storage/info",
|
||
headers=headers
|
||
)
|
||
|
||
print(f"存储信息状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("success"):
|
||
info = result["data"]
|
||
print(f"存储配额: {info['total_quota']} bytes")
|
||
print(f"已使用: {info['used_space']} bytes")
|
||
print(f"文件数量: {info['file_count']}")
|
||
|
||
except Exception as e:
|
||
print(f"获取存储信息出错: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
print("开始测试文件存储功能...")
|
||
|
||
# 1. 认证
|
||
token = test_auth()
|
||
if not token:
|
||
print("认证失败，退出测试")
|
||
exit(1)
|
||
|
||
# 2. 上传文件
|
||
file_id = test_file_upload(token)
|
||
|
||
# 3. 获取文件列表
|
||
test_file_list(token)
|
||
|
||
# 4. 获取存储信息
|
||
test_storage_info(token)
|
||
|
||
print("测试完成!") |