227 lines
8.7 KiB
Python
227 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
WordPress REST API 测试脚本
|
||
"""
|
||
|
||
import os
|
||
import requests
|
||
import json
|
||
from requests.auth import HTTPBasicAuth
|
||
from dotenv import load_dotenv
|
||
|
||
# 加载环境变量
|
||
load_dotenv()
|
||
|
||
# WordPress 配置信息
|
||
WP_URL = os.getenv('WORDPRESS_URL')
|
||
WP_USER = os.getenv('WORDPRESS_USERNAME')
|
||
WP_APP_PASSWORD = os.getenv('WORDPRESS_APP_PASSWORD')
|
||
|
||
# 移除密码中的空格
|
||
if WP_APP_PASSWORD:
|
||
WP_APP_PASSWORD = WP_APP_PASSWORD.replace(" ", "")
|
||
|
||
# 检查必需的环境变量
|
||
if not all([WP_URL, WP_USER, WP_APP_PASSWORD]):
|
||
print("❌ 错误: 请在 .env 文件中配置 WORDPRESS_URL, WORDPRESS_USERNAME, WORDPRESS_APP_PASSWORD")
|
||
exit(1)
|
||
|
||
class WordPressAPITester:
|
||
def __init__(self, url, username, app_password):
|
||
self.base_url = url.rstrip('/')
|
||
self.api_url = f"{self.base_url}/wp-json/wp/v2"
|
||
self.username = username
|
||
self.app_password = app_password
|
||
self.auth = HTTPBasicAuth(username, app_password)
|
||
|
||
def test_connection(self):
|
||
"""测试基本连接"""
|
||
print("🔍 测试 WordPress REST API 连接...")
|
||
try:
|
||
response = requests.get(f"{self.base_url}/wp-json", timeout=10)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print(f"✅ 连接成功!")
|
||
print(f" 站点名称: {data.get('name', 'N/A')}")
|
||
print(f" 描述: {data.get('description', 'N/A')}")
|
||
print(f" URL: {data.get('url', 'N/A')}")
|
||
print(f" WordPress版本: {data.get('gmt_offset', 'N/A')}")
|
||
return True
|
||
else:
|
||
print(f"❌ 连接失败: HTTP {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 连接错误: {e}")
|
||
return False
|
||
|
||
def test_authentication(self):
|
||
"""测试认证"""
|
||
print("\n🔐 测试用户认证...")
|
||
try:
|
||
response = requests.get(f"{self.api_url}/users/me", auth=self.auth, timeout=10)
|
||
if response.status_code == 200:
|
||
user_data = response.json()
|
||
print(f"✅ 认证成功!")
|
||
print(f" 用户ID: {user_data.get('id')}")
|
||
print(f" 用户名: {user_data.get('username')}")
|
||
print(f" 显示名称: {user_data.get('name')}")
|
||
print(f" 邮箱: {user_data.get('email')}")
|
||
print(f" 角色: {', '.join(user_data.get('roles', []))}")
|
||
return True
|
||
else:
|
||
print(f"❌ 认证失败: HTTP {response.status_code}")
|
||
if response.status_code == 401:
|
||
print(" 可能的原因: 用户名或应用密码错误")
|
||
elif response.status_code == 403:
|
||
print(" 可能的原因: 用户权限不足")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 认证错误: {e}")
|
||
return False
|
||
|
||
def test_posts_endpoint(self):
|
||
"""测试文章端点"""
|
||
print("\n📝 测试文章端点...")
|
||
try:
|
||
# 获取文章列表
|
||
response = requests.get(f"{self.api_url}/posts?per_page=5", timeout=10)
|
||
if response.status_code == 200:
|
||
posts = response.json()
|
||
print(f"✅ 文章端点正常! 找到 {len(posts)} 篇文章")
|
||
for i, post in enumerate(posts[:3], 1):
|
||
print(f" {i}. {post.get('title', {}).get('rendered', 'N/A')} (ID: {post.get('id')})")
|
||
return True
|
||
else:
|
||
print(f"❌ 文章端点失败: HTTP {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 文章端点错误: {e}")
|
||
return False
|
||
|
||
def test_categories_endpoint(self):
|
||
"""测试分类端点"""
|
||
print("\n📂 测试分类端点...")
|
||
try:
|
||
response = requests.get(f"{self.api_url}/categories", timeout=10)
|
||
if response.status_code == 200:
|
||
categories = response.json()
|
||
print(f"✅ 分类端点正常! 找到 {len(categories)} 个分类")
|
||
for i, cat in enumerate(categories[:3], 1):
|
||
print(f" {i}. {cat.get('name')} (ID: {cat.get('id')}, 文章数: {cat.get('count')})")
|
||
return True
|
||
else:
|
||
print(f"❌ 分类端点失败: HTTP {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 分类端点错误: {e}")
|
||
return False
|
||
|
||
def test_media_endpoint(self):
|
||
"""测试媒体端点"""
|
||
print("\n🖼️ 测试媒体端点...")
|
||
try:
|
||
response = requests.get(f"{self.api_url}/media?per_page=5", timeout=10)
|
||
if response.status_code == 200:
|
||
media = response.json()
|
||
print(f"✅ 媒体端点正常! 找到 {len(media)} 个媒体文件")
|
||
for i, item in enumerate(media[:3], 1):
|
||
print(f" {i}. {item.get('title', {}).get('rendered', 'N/A')} (类型: {item.get('media_type')})")
|
||
return True
|
||
else:
|
||
print(f"❌ 媒体端点失败: HTTP {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 媒体端点错误: {e}")
|
||
return False
|
||
|
||
def test_write_permissions(self):
|
||
"""测试写入权限"""
|
||
print("\n✏️ 测试写入权限...")
|
||
try:
|
||
# 尝试创建一个测试文章草稿
|
||
test_post = {
|
||
'title': 'API测试文章 - 请删除',
|
||
'content': '这是一个API测试文章，可以安全删除。',
|
||
'status': 'draft' # 草稿状态，不会公开发布
|
||
}
|
||
|
||
response = requests.post(
|
||
f"{self.api_url}/posts",
|
||
json=test_post,
|
||
auth=self.auth,
|
||
timeout=10
|
||
)
|
||
|
||
if response.status_code == 201:
|
||
created_post = response.json()
|
||
post_id = created_post.get('id')
|
||
print(f"✅ 写入权限正常! 创建了测试文章 (ID: {post_id})")
|
||
|
||
# 立即删除测试文章
|
||
delete_response = requests.delete(
|
||
f"{self.api_url}/posts/{post_id}?force=true",
|
||
auth=self.auth,
|
||
timeout=10
|
||
)
|
||
|
||
if delete_response.status_code == 200:
|
||
print(f"✅ 测试文章已清理 (ID: {post_id})")
|
||
else:
|
||
print(f"⚠️ 测试文章创建成功但清理失败 (ID: {post_id})，请手动删除")
|
||
|
||
return True
|
||
else:
|
||
print(f"❌ 写入权限测试失败: HTTP {response.status_code}")
|
||
if response.status_code == 401:
|
||
print(" 可能的原因: 认证失败")
|
||
elif response.status_code == 403:
|
||
print(" 可能的原因: 用户没有发布文章的权限")
|
||
return False
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 写入权限测试错误: {e}")
|
||
return False
|
||
|
||
def run_all_tests(self):
|
||
"""运行所有测试"""
|
||
print("🚀 开始 WordPress REST API 全面测试\n")
|
||
print(f"测试目标: {self.base_url}")
|
||
print(f"用户: {self.username}")
|
||
print("=" * 50)
|
||
|
||
tests = [
|
||
self.test_connection,
|
||
self.test_authentication,
|
||
self.test_posts_endpoint,
|
||
self.test_categories_endpoint,
|
||
self.test_media_endpoint,
|
||
self.test_write_permissions
|
||
]
|
||
|
||
passed = 0
|
||
total = len(tests)
|
||
|
||
for test in tests:
|
||
if test():
|
||
passed += 1
|
||
|
||
print("\n" + "=" * 50)
|
||
print(f"📊 测试结果: {passed}/{total} 项测试通过")
|
||
|
||
if passed == total:
|
||
print("🎉 所有测试通过! WordPress REST API 工作正常")
|
||
elif passed >= total * 0.8:
|
||
print("⚠️ 大部分测试通过，但有一些问题需要注意")
|
||
else:
|
||
print("❌ 多项测试失败，请检查配置和权限")
|
||
|
||
return passed, total
|
||
|
||
def main():
|
||
"""主函数"""
|
||
tester = WordPressAPITester(WP_URL, WP_USER, WP_APP_PASSWORD)
|
||
tester.run_all_tests()
|
||
|
||
if __name__ == "__main__":
|
||
main() |