cccb7cfc
 tangwang
init
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
#!/usr/bin/env python3
"""
OAuth2.0 认证测试脚本
用于验证Shoplazza OAuth流程是否正常工作
"""
import requests
import json
import os
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
def test_health_check(base_url):
"""测试健康检查端点"""
try:
response = requests.get(f"{base_url}/health")
if response.status_code == 200:
print("✅ 健康检查通过")
print(f" 状态: {response.json()}")
return True
else:
print(f"❌ 健康检查失败: {response.status_code}")
return False
except Exception as e:
print(f"❌ 健康检查异常: {str(e)}")
return False
def test_auth_endpoints(base_url):
"""测试认证端点"""
print("\n🔐 测试认证端点...")
# 测试根端点
try:
response = requests.get(base_url)
if response.status_code == 200:
print("✅ 根端点正常")
endpoints = response.json().get('endpoints', {})
print(f" 可用端点: {list(endpoints.keys())}")
else:
print(f"❌ 根端点异常: {response.status_code}")
except Exception as e:
print(f"❌ 根端点异常: {str(e)}")
def test_oauth_flow(base_url, shop_domain):
"""测试OAuth流程"""
print(f"\n🔄 测试OAuth流程 (商店: {shop_domain})...")
# 构建认证URL
auth_url = f"{base_url}/auth/install?shop={shop_domain}"
print(f"认证URL: {auth_url}")
# 测试认证端点（不跟随重定向）
try:
response = requests.get(auth_url, allow_redirects=False)
if response.status_code in [302, 301]:
print("✅ 认证重定向正常")
print(f" 重定向到: {response.headers.get('Location', 'N/A')}")
else:
print(f"❌ 认证重定向异常: {response.status_code}")
except Exception as e:
print(f"❌ 认证端点异常: {str(e)}")
def test_api_endpoints(base_url, shop_domain):
"""测试API端点"""
print(f"\n📡 测试API端点 (商店: {shop_domain})...")
endpoints = [
f"/api/customers/{shop_domain}",
f"/api/products/{shop_domain}",
f"/api/orders/{shop_domain}",
f"/api/shop_info/{shop_domain}"
]
for endpoint in endpoints:
try:
response = requests.get(f"{base_url}{endpoint}")
if response.status_code == 401:
print(f"⚠️ {endpoint} - 需要认证 (正常)")
elif response.status_code == 200:
print(f"✅ {endpoint} - 认证成功")
else:
print(f"❌ {endpoint} - 异常状态: {response.status_code}")
except Exception as e:
print(f"❌ {endpoint} - 异常: {str(e)}")
def test_tokens_endpoint(base_url):
"""测试令牌端点"""
print("\n🔑 测试令牌端点...")
try:
response = requests.get(f"{base_url}/auth/tokens")
if response.status_code == 200:
tokens = response.json()
print("✅ 令牌端点正常")
print(f" 已授权商店: {tokens.get('authorized_shops', [])}")
if tokens.get('tokens'):
print(f" 令牌数量: {len(tokens.get('tokens', {}))}")
else:
print(f"❌ 令牌端点异常: {response.status_code}")
except Exception as e:
print(f"❌ 令牌端点异常: {str(e)}")
def main():
"""主测试函数"""
print("🚀 Shoplazza OAuth2.0 认证测试")
print("=" * 50)
# 获取配置
base_url = os.getenv('BASE_URL', 'http://localhost:3000')
shop_domain = input("请输入商店域名 (例如: your-shop.myshoplaza.com): ").strip()
if not shop_domain:
shop_domain = "your-shop.myshoplaza.com"
print(f"使用默认商店域名: {shop_domain}")
print(f"\n测试配置:")
print(f" 基础URL: {base_url}")
print(f" 商店域名: {shop_domain}")
# 执行测试
if test_health_check(base_url):
test_auth_endpoints(base_url)
test_oauth_flow(base_url, shop_domain)
test_api_endpoints(base_url, shop_domain)
test_tokens_endpoint(base_url)
print("\n" + "=" * 50)
print("🎯 测试完成!")
print("\n下一步操作:")
print(f"1. 在浏览器中访问: {base_url}/auth/install?shop={shop_domain}")
print("2. 完成OAuth认证流程")
print("3. 重新运行此测试脚本验证API调用")
if __name__ == "__main__":
main()
|