#!/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()